issue #45: Refactor cache locking mechanism and update dependencies f…

CLAUDE.md

@@ -21,10 +21,11 @@ The library is organized into several core modules:
 ### Key Components
 
 1. **ReqShield**: Main orchestrator that manages cache operations and request collapsing
-2. **KeyLock**: Locking mechanism (local or global) to prevent concurrent cache operations
+2. **KeyLock**: Locking mechanism (local or global) to prevent concurrent cache operations. `tryLock` returns an ownership token and `unLock` only releases when the token matches, so a slow holder cannot release someone else's lock
 3. **ReqShieldConfiguration**: Configuration object that defines cache functions, locking behavior, and timeouts
 4. **ReqShieldData**: Wrapper for cached data with metadata (creation time, TTL)
-5. **Spring Aspects**: AOP-based implementations that provide annotation-driven caching
+5. **Spring Aspects**: AOP-based implementations that provide annotation-driven caching. One `ReqShield` instance per annotated method; cache and lock keys are namespaced as `"{cacheName}::{key}"`; eviction happens after the method returns successfully
+6. **GlobalLockSupport**: Optional interface a cache bean implements to enable `isLocalLock = false` (Redis `SET NX PX` + compare-and-delete). Without it, `isLocalLock = false` fails fast
 
 ### Design Patterns
 
@@ -88,29 +89,31 @@ Contains shared:
 
 ### ReqShieldConfiguration Parameters
 - `isLocalLock`: Use local vs distributed locking (default: true)
+- `globalLockFunction` / `globalUnLockFunction`: `(lockKey, token, ttlMillis) -> Boolean` / `(lockKey, token) -> Boolean`, required when `isLocalLock = false`
+- `executor` (core) / `scheduler` (reactor) / `scope` (coroutine): where background cache writes run; defaults are shared, the Spring adapters expose them as `reqShieldExecutor` / `reqShieldScheduler` / `reqShieldCoroutineScope` beans
 - `lockTimeoutMillis`: Lock acquisition timeout (default: 3000ms)
 - `decisionForUpdate`: Percentage of TTL after which to trigger async cache refresh (default: 80)
-- `maxAttemptGetCache`: Max retry attempts when waiting for cache (default: 60)
+- `maxAttemptGetCache`: Max retry attempts when waiting for cache (default: 60, 50ms apart). Three consecutive cache-read failures while waiting fall back to the supplier immediately; supplier failures propagate as `ClientException(SUPPLIER_ERROR)`
 - `reqShieldWorkMode`: CREATE_AND_UPDATE_CACHE | ONLY_CREATE_CACHE | ONLY_UPDATE_CACHE
 
 ### Work Modes
 - **CREATE_AND_UPDATE_CACHE**: Full functionality (default)
-- **ONLY_CREATE_CACHE**: Never updates existing cache entries
-- **ONLY_UPDATE_CACHE**: Never creates new cache entries
+- **ONLY_CREATE_CACHE**: Request collapsing (lock) applies only when a cache entry is being created; refreshes of existing entries still happen but without a lock, so every request past the `decisionForUpdate` threshold may trigger a refresh
+- **ONLY_UPDATE_CACHE**: Request collapsing (lock) applies only when an existing entry is being refreshed; on a cache miss every request calls the supplier and writes the cache without a lock
 
 ## Testing Guidelines
 
 ### Test Infrastructure
 - Uses JUnit 5 platform
 - MockK for Kotlin mocking
-- Testcontainers for integration tests (Redis)
+- Testcontainers for integration tests (Redis); set `TEST_REDIS_HOST`/`TEST_REDIS_PORT` to use an external Redis instead (CI does this)
 - Awaitility for asynchronous testing
 - Separate test fixtures in `support` module
 
 ### Test Coverage Requirements
-- **Minimum test coverage**: 80% must be maintained across all modules
+- **Minimum test coverage**: 80% line coverage for every library module (example modules are exempt)
 - Coverage reports generated via `./gradlew jacocoTestReport`
-- Coverage enforced through Jacoco plugin configuration
+- Enforced by `jacocoTestCoverageVerification`, which runs as part of `./gradlew check` / `./gradlew build`
 
 ### Code Quality Requirements
 - **Lint validation**: All code must pass ktlint checks before completion

README.md

@@ -39,27 +39,59 @@ A lib that regulates the cache-based requests an application receives in terms o
   - `EMIT_EMPTY` (default): map `null` to `Mono.empty()`.
   - `ERROR`: throw an `IllegalStateException` if a `null` value is produced.
 
-### Global lock guidance
-
-- When `isLocalLock = false`, you must provide real global lock/unlock implementations.
-- Recommended approach with Redis:
-  - Lock: `SETNX lock:{key} 1` + `PEXPIRE lock:{key} {ttlMillis}`
-  - Unlock: `DEL lock:{key}`
-- The provided defaults return `true` and are only suitable for local/dev usage.
-
-### Reactor Scheduler tuning
+### Cache key layout
 
-- Reactor-based modules accept a `Scheduler` (e.g., `boundedElastic`) through configuration.
-- Spring WebFlux adapter exposes a `reqShieldScheduler` bean you can override for tuning thread usage.
+- The Spring adapters store every entry under `"{cacheName}::{key}"` (the same convention as Spring's `RedisCacheManager`),
+  where `key` is the SpEL result or the `KeyGenerator` output. `@ReqShieldCacheEvict` applies the same rule, so an evict
+  with the same `cacheName` and `key` always targets the entry written by `@ReqShieldCacheable`.
+- Lock keys are derived from that namespaced key and prefixed with `reqshield:lock:`, so two caches that happen to use the
+  same raw key never share a lock or an entry.
+- If you call `ReqShield` directly (core / core-reactor / core-kotlin-coroutine), the key you pass is used as is.
 
-### Kotlin Coroutine Parallelism Configuration
-
-| Property | Default | Description |
-|----------|---------|-------------|
-| `reqshield.blocking.parallelism` | `availableProcessors * 2` (clamped 4-256) | Controls parallelism for blocking calls in the coroutine aspect |
+### Global lock guidance
 
-**Note**: This feature uses `Dispatchers.IO.limitedParallelism()` which is marked as `@ExperimentalCoroutinesApi`.
-The API may change in future Kotlin Coroutines versions.
+- When `isLocalLock = false`, your cache bean must also implement the module's `GlobalLockSupport` interface
+  (`com.linecorp.cse.reqshield.spring.cache.GlobalLockSupport`, `...spring.webflux.cache.GlobalLockSupport`,
+  `...spring.webflux.kotlin.coroutine.cache.GlobalLockSupport`). If it does not, the first call to the annotated method
+  fails with an `IllegalArgumentException` instead of silently running without request collapsing.
+- Every lock acquisition carries an ownership token. Only the holder that acquired the lock can release it, so a slow
+  holder whose lock already expired can no longer release the lock of the next holder.
+- Recommended Redis implementation:
+  - Lock: `SET {lockKey} {token} NX PX {ttlMillis}` (atomic; never `SETNX` followed by a separate `PEXPIRE`)
+  - Unlock: compare-and-delete in a Lua script, e.g.
+    `if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end`
+- The example modules contain working implementations for `RedisTemplate`, `ReactiveRedisTemplate` and the coroutine
+  extensions.
+
+### Cache eviction semantics
+
+- `@ReqShieldCacheEvict` evicts **after** the annotated method completes successfully (Spring's `@CacheEvict` default).
+  If the method throws, nothing is evicted; if the eviction itself fails, that failure propagates to the caller.
+- In WebFlux the eviction also runs when the method completes empty (for example a `Mono<Void>` handler).
+
+### Waiting for another request (lock not acquired)
+
+- A request that loses the lock polls the cache every 50 ms, up to `maxAttemptGetCache` times (default 60).
+- A cache read failure while polling is logged and counted as a failed attempt. Three consecutive failures are treated as a
+  cache outage and the request falls back to calling the supplier itself right away.
+- When the attempts are exhausted the request calls the supplier once. A supplier failure is propagated as a
+  `ClientException(SUPPLIER_ERROR)` with the original exception as `cause`; it is never turned into a cached `null`.
+
+### Thread pools and schedulers
+
+- `core` uses a `ScheduledExecutorService`. The default is a shared daemon pool; the Spring adapter exposes it as the
+  `reqShieldExecutor` bean, which you can override.
+- `core-reactor` accepts a `Scheduler` (default `boundedElastic`). The Spring WebFlux adapter exposes it as the
+  `reqShieldScheduler` bean.
+- `core-kotlin-coroutine` accepts a `CoroutineScope` for background cache writes (default: a shared supervisor scope on
+  `Dispatchers.IO`). The coroutine Spring adapter exposes it as the `reqShieldCoroutineScope` bean and cancels it on
+  context shutdown.
+
+### Kotlin coroutine adapter
+
+- `@ReqShieldCacheable` / `@ReqShieldCacheEvict` from `core-spring-webflux-kotlin-coroutine` require `suspend`
+  functions. Annotating a regular function fails fast with an `IllegalArgumentException`; use `core-spring` or
+  `core-spring-webflux` for blocking or `Mono`-returning methods.
 
 ## Contributing
 

build.gradle.kts

@@ -63,6 +63,26 @@ allprojects {
         }
     }
 
+    // Enforce the minimum line coverage documented in CLAUDE.md for library modules.
+    // Example applications (req-shield-*-example) are demos and are not held to the threshold.
+    if (project != rootProject && !project.name.startsWith("req-shield-")) {
+        tasks.withType<JacocoCoverageVerification> {
+            dependsOn(tasks.test)
+            violationRules {
+                rule {
+                    limit {
+                        counter = "LINE"
+                        value = "COVEREDRATIO"
+                        minimum = "0.80".toBigDecimal()
+                    }
+                }
+            }
+        }
+        tasks.named("check") {
+            dependsOn(tasks.withType<JacocoCoverageVerification>())
+        }
+    }
+
     jacoco {
         toolVersion = "0.8.12"
     }

core-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/KeyGlobalLock.kt

@@ -16,24 +16,25 @@
 
 package com.linecorp.cse.reqshield.kotlin.coroutine
 
+import java.util.UUID
+
 class KeyGlobalLock(
-    private val globalLockFunction: suspend (String, Long) -> Boolean,
-    private val globalUnLockFunction: suspend (String) -> Boolean,
+    private val globalLockFunction: suspend (String, String, Long) -> Boolean,
+    private val globalUnLockFunction: suspend (String, String) -> Boolean,
     private val lockTimeoutMillis: Long,
 ) : KeyLock {
     override suspend fun tryLock(
         key: String,
         lockType: LockType,
-    ): Boolean {
-        val completeKey = "${key}_${lockType.name}"
-        return globalLockFunction(completeKey, lockTimeoutMillis)
+    ): String? {
+        // The token must be unique across every process sharing the lock store.
+        val token = UUID.randomUUID().toString()
+        return if (globalLockFunction(lockKeyOf(key, lockType), token, lockTimeoutMillis)) token else null
     }
 
     override suspend fun unLock(
         key: String,
         lockType: LockType,
-    ): Boolean {
-        val completeKey = "${key}_${lockType.name}"
-        return globalUnLockFunction(completeKey)
-    }
+        token: String,
+    ): Boolean = globalUnLockFunction(lockKeyOf(key, lockType), token)
 }

core-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/KeyLocalLock.kt

@@ -18,6 +18,7 @@ package com.linecorp.cse.reqshield.kotlin.coroutine
 
 import com.linecorp.cse.reqshield.support.constant.ConfigValues.LOCK_MONITOR_INTERVAL_MILLIS
 import com.linecorp.cse.reqshield.support.utils.nowToEpochTime
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -28,6 +29,8 @@ import org.slf4j.LoggerFactory
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.Semaphore
 import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicLong
+import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.CoroutineContext
 
 private val log = LoggerFactory.getLogger(KeyLocalLock::class.java)
@@ -51,22 +54,36 @@ class KeyLocalLock(private val lockTimeoutMillis: Long) : KeyLock, CoroutineScop
          * check vs unLock, or monitor cleanup vs unLock).
          */
         val isHeld: AtomicBoolean = AtomicBoolean(false),
+        /**
+         * Token of the current owner, or null when the lock is not held.
+         * Only the owner that presents this exact token may release the lock, so a holder whose
+         * lock already expired and was reacquired by somebody else cannot release the new owner.
+         */
+        @Volatile var token: String? = null,
     )
 
     companion object {
         private val lockMap = ConcurrentHashMap<String, LockInfo>()
 
+        /**
+         * Source of local ownership tokens. A monotonic counter is enough because tokens never
+         * leave this JVM, and it is far cheaper than a UUID on the lock acquisition path.
+         */
+        private val tokenSequence = AtomicLong()
+
         @Volatile
         private var monitorJob: Job? = null
 
+        private fun nextToken(): String = "local-${tokenSequence.incrementAndGet()}"
+
         private fun ensureMonitorStarted() {
             if (monitorJob?.isActive == true) return
             synchronized(this) {
                 if (monitorJob?.isActive == true) return
                 monitorJob =
                     CoroutineScope(Dispatchers.IO).launch {
                         while (isActive) {
-                            runCatching {
+                            try {
                                 val now = System.currentTimeMillis()
                                 // Remove expired locks using compute() for atomic check-and-remove.
                                 // This prevents TOCTOU race condition where removeIf's lambda returns true
@@ -82,6 +99,7 @@ class KeyLocalLock(private val lockTimeoutMillis: Long) : KeyLock, CoroutineScop
                                             // This handles the case where unlock() was missed due to exception.
                                             // CAS ensures safe release (no-op if already released).
                                             if (lockInfo.isHeld.compareAndSet(true, false)) {
+                                                lockInfo.token = null
                                                 lockInfo.semaphore.release()
                                             }
                                             null // Atomic removal
@@ -90,10 +108,15 @@ class KeyLocalLock(private val lockTimeoutMillis: Long) : KeyLock, CoroutineScop
                                         }
                                     }
                                 }
-                                delay(LOCK_MONITOR_INTERVAL_MILLIS)
-                            }.onFailure { e ->
+                            } catch (e: CancellationException) {
+                                throw e
+                            } catch (e: Exception) {
                                 log.error("Error in lock lifecycle monitoring: {}", e.message, e)
                             }
+
+                            // Delay outside the try/catch: a failed sweep must still wait for the next
+                            // interval instead of turning the monitor into a hot loop.
+                            delay(LOCK_MONITOR_INTERVAL_MILLIS)
                         }
                     }
             }
@@ -119,57 +142,74 @@ class KeyLocalLock(private val lockTimeoutMillis: Long) : KeyLock, CoroutineScop
     override suspend fun tryLock(
         key: String,
         lockType: LockType,
-    ): Boolean {
-        val completeKey = "${key}_${lockType.name}"
+    ): String? {
+        val completeKey = lockKeyOf(key, lockType)
         val now = nowToEpochTime()
-        val result = AtomicBoolean(false)
+        val acquiredToken = AtomicReference<String?>(null)
 
         // Use compute() for atomic lock acquisition.
-        // This ensures mutual exclusion with cleanup - they cannot race on the same key.
+        // This ensures mutual exclusion with cleanup and unLock - they cannot race on the same key.
         lockMap.compute(completeKey) { _, existing ->
             if (existing != null) {
                 // Force-release expired locks to allow reacquisition.
                 // Use CAS to prevent race condition with concurrent unLock().
                 // Without CAS, if unLock() executes between isHeld.get() and release(),
                 // both threads would call release(), causing over-release (permits > 1).
                 if (now > existing.expiresAt && existing.isHeld.compareAndSet(true, false)) {
+                    // Drop the token as well so the timed-out owner cannot release the next one.
+                    existing.token = null
                     existing.semaphore.release()
                 }
 
                 // Existing entry: try to acquire semaphore
                 if (existing.semaphore.tryAcquire()) {
+                    val token = nextToken()
                     existing.isHeld.set(true)
+                    existing.token = token
                     existing.expiresAt = now + lockTimeoutMillis
-                    result.set(true)
+                    acquiredToken.set(token)
                 }
                 existing
             } else {
                 // New entry: create and acquire
-                val newLock = LockInfo(Semaphore(1), now + lockTimeoutMillis)
+                val token = nextToken()
+                val newLock = LockInfo(Semaphore(1), now + lockTimeoutMillis, token = token)
                 newLock.semaphore.tryAcquire() // Always succeeds for new semaphore
                 newLock.isHeld.set(true)
-                result.set(true)
+                acquiredToken.set(token)
                 newLock
             }
         }
-        return result.get()
+        return acquiredToken.get()
     }
 
     override suspend fun unLock(
         key: String,
         lockType: LockType,
+        token: String,
     ): Boolean {
-        val completeKey = "${key}_${lockType.name}"
-        val lockInfo = lockMap[completeKey] ?: return false
-
-        // Use CAS to prevent over-release: only release if we actually hold the lock
-        return if (lockInfo.isHeld.compareAndSet(true, false)) {
-            lockInfo.semaphore.release()
-            true
-        } else {
-            log.debug("Attempted to unlock key '{}' that is not held", completeKey)
-            false
+        val completeKey = lockKeyOf(key, lockType)
+        val released = AtomicBoolean(false)
+
+        // Release inside compute() so that it is atomic with respect to acquisition and cleanup
+        // of the same key: the token check and the semaphore release cannot be interleaved with a
+        // reacquisition. The entry itself is kept so a waiting caller can still acquire it.
+        lockMap.compute(completeKey) { _, existing ->
+            if (existing == null) return@compute null
+
+            // Only the current owner may release: a stale token belongs to an expired holder.
+            if (existing.token == token && existing.isHeld.compareAndSet(true, false)) {
+                existing.token = null
+                existing.semaphore.release()
+                released.set(true)
+            }
+            existing
+        }
+
+        if (!released.get()) {
+            log.debug("Attempted to unlock key '{}' that is not held by token '{}'", completeKey, token)
         }
+        return released.get()
     }
 
     fun cancel() {

core-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/KeyLock.kt

@@ -16,19 +16,30 @@
 
 package com.linecorp.cse.reqshield.kotlin.coroutine
 
+import com.linecorp.cse.reqshield.support.constant.ConfigValues.LOCK_KEY_PREFIX
+
 interface KeyLock {
+    /** Returns an opaque ownership token when acquired, or null when another holder owns the lock. */
     suspend fun tryLock(
         key: String,
         lockType: LockType,
-    ): Boolean
+    ): String?
 
+    /** Releases only if [token] matches the current owner; false when not held or token mismatch. */
     suspend fun unLock(
         key: String,
         lockType: LockType,
+        token: String,
     ): Boolean
 }
 
 enum class LockType {
     CREATE,
     UPDATE,
 }
+
+/** Lock keys are prefixed so that a lock entry can never collide with a cache entry. */
+internal fun lockKeyOf(
+    key: String,
+    lockType: LockType,
+): String = "$LOCK_KEY_PREFIX${key}_${lockType.name}"