diff --git a/CLAUDE.md b/CLAUDE.md index 3d0147e..60cf992 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -90,7 +90,7 @@ 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 +- `executor` (core) / `scheduler` (reactor) / `scope` (coroutine): where background cache writes run; defaults are shared. The Spring adapters register none of them as beans (an `Executor` bean would make Spring Boot drop its `applicationTaskExecutor`, and any library bean would clash with an application bean of the same name): each aspect uses an application bean named `reqShieldExecutor` / `reqShieldScheduler` / `reqShieldCoroutineScope` if one exists (a bean of that name with another type fails the context refresh), otherwise its own default (an aspect-owned pool / the shared `boundedElastic` / an aspect-owned scope), and only shuts down what it owns - `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, 50ms apart). Three consecutive cache-read failures while waiting fall back to the supplier immediately; supplier failures propagate as `ClientException(SUPPLIER_ERROR)` diff --git a/README.md b/README.md index 270bcfd..2dbb615 100644 --- a/README.md +++ b/README.md @@ -155,13 +155,27 @@ refresh dependencies with `./gradlew build --refresh-dependencies`. ### Thread pools and schedulers - `core` accepts any `java.util.concurrent.Executor` for its background cache writes; only `execute` is called and the - library never shuts a caller-supplied pool down. The default is a shared daemon pool; the Spring adapter exposes it as - the `reqShieldExecutor` bean (an `ExecutorService` the context shuts down), which you can override. -- `core-reactor` accepts a `Scheduler` (default `boundedElastic`). The Spring WebFlux adapter exposes it as the - `reqShieldScheduler` bean. + library never shuts a caller-supplied pool down. The default is a shared daemon pool. The Spring adapter runs the + writes on a daemon pool of its own that is shut down with the context, and registers no `Executor` bean, so Spring + Boot keeps its `applicationTaskExecutor`. To use your own pool, define an `Executor` bean named `reqShieldExecutor`; + its lifecycle stays yours, and like any `Executor` bean it makes Spring Boot skip `applicationTaskExecutor`. +- `core-reactor` accepts a `Scheduler` (default `boundedElastic`). The Spring WebFlux adapter uses the shared + `boundedElastic` too and never disposes it; to use your own, define a `Scheduler` bean named `reqShieldScheduler`. - `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. + `Dispatchers.IO`). The coroutine Spring adapter runs them on a supervisor scope of its own on `Dispatchers.IO` and + cancels it on context shutdown; to use your own, define a `CoroutineScope` bean named `reqShieldCoroutineScope`. +- The Spring adapters register none of these beans themselves, so an application bean of the same name replaces the + default instead of clashing with it, and its lifecycle stays with the application. A bean of one of these names + that is not of the expected type fails the application startup instead of being ignored. + +### WebFlux adapter + +- `@ReqShieldCacheable` / `@ReqShieldCacheEvict` from `core-spring-webflux` require methods that return `Mono`. The + aspect hands back a lazy `Mono`, and other return types are not rejected up front: + - On a `void` (`Unit`) method Spring discards that `Mono`, so the method body never runs and nothing is evicted, + without any error. Return `Mono` from eviction methods instead. + - Any other return type fails with a `ClassCastException` at the call site. +- Use `core-spring` for blocking methods and `core-spring-webflux-kotlin-coroutine` for `suspend` functions. ### Kotlin coroutine adapter diff --git a/core-reactor/src/main/kotlin/com/linecorp/cse/reqshield/reactor/ReqShield.kt b/core-reactor/src/main/kotlin/com/linecorp/cse/reqshield/reactor/ReqShield.kt index 42ca3e8..9827876 100644 --- a/core-reactor/src/main/kotlin/com/linecorp/cse/reqshield/reactor/ReqShield.kt +++ b/core-reactor/src/main/kotlin/com/linecorp/cse/reqshield/reactor/ReqShield.kt @@ -151,6 +151,7 @@ class ReqShield( }, ).doFinally { // A cache hit, read failure, or cancellation during the recheck must release our token. + // Once creation started, createReqShieldData releases it (error, cancel or after the write). if (!cacheCreationStarted.get()) { releaseLock(key, lockType, token) } @@ -169,10 +170,14 @@ class ReqShield( timeToLiveMillis: Long, lockType: LockType, token: String?, - ): Mono> = - executeCallable({ callable.call() }, key, lockType, token) + ): Mono> { + // Set once the lock is handed to the asynchronous cache write, which releases it after writing + val lockHandedToCacheWrite = AtomicBoolean(false) + + return executeCallable({ callable.call() }, key, lockType, token) .map { data -> buildReqShieldData(data, timeToLiveMillis) } .doOnNext { reqShieldData -> + lockHandedToCacheWrite.set(true) // Async fire-and-forget cache storage (matches coroutine implementation) setReqShieldData( reqShieldConfig.setCacheFunction, @@ -180,13 +185,13 @@ class ReqShield( reqShieldData, lockType, token, - ).subscribeOn(reqShieldConfig.scheduler) - .subscribe( - { /* success - no action needed */ }, - { e -> log.error("Failed to set cache for key '{}': {}", key, e.message, e) }, - ) + ).subscribe( + { /* success - no action needed */ }, + { e -> log.error("Failed to set cache for key '{}': {}", key, e.message, e) }, + ) }.switchIfEmpty( Mono.defer { + lockHandedToCacheWrite.set(true) val reqShieldData = buildReqShieldData(null, timeToLiveMillis) // Async fire-and-forget cache storage (matches coroutine implementation) setReqShieldData( @@ -195,14 +200,20 @@ class ReqShield( reqShieldData, lockType, token, - ).subscribeOn(reqShieldConfig.scheduler) - .subscribe( - { /* success - no action needed */ }, - { e -> log.error("Failed to set cache for key '{}': {}", key, e.message, e) }, - ) + ).subscribe( + { /* success - no action needed */ }, + { e -> log.error("Failed to set cache for key '{}': {}", key, e.message, e) }, + ) Mono.just(reqShieldData) }, - ) + ).doOnCancel { + // A caller cancelled while the supplier runs (client disconnect, timeout()) never reaches + // the cache write, so the lock is released here instead of lingering until it expires. + if (token != null && !lockHandedToCacheWrite.get()) { + releaseLock(key, lockType, token) + } + } + } /** * Waits for the request that owns the lock to fill the cache. @@ -303,12 +314,15 @@ class ReqShield( Mono .defer { setFunction(key, value, value.timeToLiveMillis) } .onErrorMap { e -> ClientException(ErrorCode.SET_CACHE_ERROR, cause = e) } + .subscribeOn(reqShieldConfig.scheduler) .doFinally { // Only the holder of a token took a lock, so only it may release one. + // Placed after subscribeOn: a scheduler that rejects the write (disposed or saturated) never + // subscribes the operators above it, so a release there would never run. if (token != null) { releaseLock(key, lockType, token) } - }.subscribeOn(reqShieldConfig.scheduler) + } private fun releaseLock( key: String, diff --git a/core-reactor/src/test/kotlin/com/linecorp/cse/reqshield/reactor/ReqShieldTest.kt b/core-reactor/src/test/kotlin/com/linecorp/cse/reqshield/reactor/ReqShieldTest.kt index 68240f4..b7119cf 100644 --- a/core-reactor/src/test/kotlin/com/linecorp/cse/reqshield/reactor/ReqShieldTest.kt +++ b/core-reactor/src/test/kotlin/com/linecorp/cse/reqshield/reactor/ReqShieldTest.kt @@ -29,19 +29,25 @@ import io.mockk.every import io.mockk.mockk import io.mockk.slot import io.mockk.verify +import org.awaitility.Awaitility.await import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertThrows +import reactor.core.Disposable import reactor.core.publisher.Mono import reactor.core.publisher.MonoSink +import reactor.core.scheduler.Scheduler import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import java.lang.reflect.InvocationTargetException import java.lang.reflect.Method import java.time.Duration import java.util.concurrent.Callable +import java.util.concurrent.RejectedExecutionException +import java.util.concurrent.TimeoutException +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicReference import kotlin.test.assertNotNull @@ -252,6 +258,62 @@ class ReqShieldTest : BaseReqShieldTest { verify(exactly = 0) { callable.call() } } + @Test + fun shouldReleaseLockWhenCallerCancelsWhileSupplierRuns() { + every { cacheGetter(key) } returns Mono.empty() + every { keyLock.tryLock(key, LockType.CREATE) } returns Mono.just(token) + every { keyLock.unLock(key, LockType.CREATE, token) } returns Mono.just(true) + every { callable.call() } returns Mono.never() + + StepVerifier.create(reqShield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + .then { verify(timeout = 1000, exactly = 1) { callable.call() } } + .thenCancel() + .verify() + + verify(timeout = 1000, exactly = 1) { keyLock.unLock(key, LockType.CREATE, token) } + verify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + @Test + fun shouldFreeLocalLockForNextRequestWhenCallerTimesOutWhileSupplierRuns() { + val cancelKey = "cancel-${System.nanoTime()}" + val localLock = KeyLocalLock(60_000) + val shield = ReqShield(ReqShieldConfiguration({ _, _, _ -> Mono.just(true) }, { Mono.empty() }, keyLock = localLock)) + + StepVerifier.create( + shield + .getAndSetReqShieldData(cancelKey, Callable { Mono.never() }, timeToLiveMillis) + .timeout(Duration.ofMillis(100)), + ).expectError(TimeoutException::class.java) + .verify() + + // Without the release, the lock would stay held for its whole 60s timeout + await().atMost(Duration.ofSeconds(1)).until { localLock.tryLock(cancelKey, LockType.CREATE).block() != null } + } + + @Test + fun shouldReleaseLockOnlyAfterCacheWriteWhenCallerCancelsAfterValueIsEmitted() { + lateinit var pendingWrite: MonoSink + every { cacheGetter(key) } returns Mono.empty() + every { cacheSetter(key, any(), any()) } returns Mono.create { pendingWrite = it } + every { keyLock.tryLock(key, LockType.CREATE) } returns Mono.just(token) + every { keyLock.unLock(key, LockType.CREATE, token) } returns Mono.just(true) + val shield = + ReqShield( + ReqShieldConfiguration(cacheSetter, cacheGetter, keyLock = keyLock, scheduler = Schedulers.immediate()), + ) + + StepVerifier.create(shield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + .expectNextMatches { it.value == value } + .thenCancel() + .verify() + + // The write already owns the lock, so the cancellation must not release it early + verify(exactly = 0) { keyLock.unLock(key, LockType.CREATE, token) } + pendingWrite.success(true) + verify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, token) } + } + @Test fun shouldKeepLockUntilAsyncCacheWriteCompletesAfterRecheckMiss() { lateinit var pendingWrite: MonoSink @@ -275,6 +337,66 @@ class ReqShieldTest : BaseReqShieldTest { verify(exactly = 1) { callable.call() } } + @Test + fun shouldReturnComputedDataAndReleaseLockWhenSchedulerRejectsTheCacheWrite() { + val scheduler = RejectingScheduler() + every { cacheGetter(key) } returns Mono.empty() + every { keyLock.tryLock(key, LockType.CREATE) } returns Mono.just(token) + every { keyLock.unLock(key, LockType.CREATE, token) } returns Mono.just(true) + // Built outside answers {}, whose scope has its own `value` + val supplied = Mono.just(value) + // The supplier runs right before the write is scheduled, so only the write is rejected + every { callable.call() } answers { + scheduler.rejecting.set(true) + supplied + } + val shield = ReqShield(ReqShieldConfiguration(cacheSetter, cacheGetter, keyLock = keyLock, scheduler = scheduler)) + + StepVerifier.create(shield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + .expectNextMatches { it.value == value } + .verifyComplete() + + verify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, token) } + verify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + @Test + fun shouldReturnCachedDataAndReleaseLockWhenSchedulerRejectsTheRefreshWrite() { + val scheduler = RejectingScheduler() + val cached = updateTargetReqShieldData(oldValue) + every { cacheGetter(key) } returns Mono.just(cached) + every { keyLock.tryLock(key, LockType.UPDATE) } returns Mono.just(token) + every { keyLock.unLock(key, LockType.UPDATE, token) } returns Mono.just(true) + val supplied = Mono.just(value) + every { callable.call() } answers { + scheduler.rejecting.set(true) + supplied + } + val shield = ReqShield(ReqShieldConfiguration(cacheSetter, cacheGetter, keyLock = keyLock, scheduler = scheduler)) + + StepVerifier.create(shield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + .expectNext(cached) + .verifyComplete() + + verify(exactly = 1) { keyLock.unLock(key, LockType.UPDATE, token) } + verify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + /** Runs tasks inline until [rejecting] is set, then rejects them the way a disposed or saturated scheduler does. */ + private class RejectingScheduler : Scheduler by Schedulers.immediate() { + val rejecting = AtomicBoolean(false) + + override fun createWorker(): Scheduler.Worker { + val delegate = Schedulers.immediate().createWorker() + return object : Scheduler.Worker by delegate { + override fun schedule(task: Runnable): Disposable { + if (rejecting.get()) throw RejectedExecutionException("rejected by test") + return delegate.schedule(task) + } + } + } + } + /** Cached entry that has passed the decisionForUpdate threshold (90% of its TTL). */ private fun updateTargetReqShieldData(cachedValue: Product?): ReqShieldData = ReqShieldData( diff --git a/core-spring-webflux-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspect.kt b/core-spring-webflux-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspect.kt index 4fe44f1..600da4f 100644 --- a/core-spring-webflux-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspect.kt +++ b/core-spring-webflux-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspect.kt @@ -22,25 +22,31 @@ import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.annotation.Req import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.annotation.ReqShieldCacheable import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.cache.AsyncCache import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.cache.GlobalLockSupport +import kotlinx.coroutines.CoroutineExceptionHandler +import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel import kotlinx.coroutines.reactor.awaitSingleOrNull import org.aspectj.lang.ProceedingJoinPoint import org.aspectj.lang.annotation.Around import org.aspectj.lang.annotation.Aspect import org.aspectj.lang.reflect.MethodSignature +import org.slf4j.LoggerFactory import org.springframework.aop.support.AopUtils import org.springframework.beans.factory.BeanFactory import org.springframework.beans.factory.BeanFactoryAware -import org.springframework.beans.factory.annotation.Qualifier +import org.springframework.beans.factory.DisposableBean import org.springframework.cache.interceptor.KeyGenerator import org.springframework.cache.interceptor.SimpleKeyGenerator import org.springframework.context.expression.MethodBasedEvaluationContext import org.springframework.core.DefaultParameterNameDiscoverer -import org.springframework.core.SpringVersion import org.springframework.core.annotation.AnnotationUtils import org.springframework.expression.EvaluationContext import org.springframework.expression.Expression import org.springframework.expression.spel.standard.SpelExpressionParser +import org.springframework.util.ClassUtils import org.springframework.util.StringUtils import org.springframework.util.function.SingletonSupplier import reactor.core.publisher.Mono @@ -51,10 +57,34 @@ import kotlin.coroutines.Continuation @Aspect open class ReqShieldAspect( private val asyncCache: AsyncCache, - @Qualifier("reqShieldCoroutineScope") private val scope: CoroutineScope, -) : BeanFactoryAware { +) : BeanFactoryAware, + DisposableBean { private lateinit var beanFactory: BeanFactory - private val springVersion = SpringVersion.getVersion() + + /** Set only when no `reqShieldCoroutineScope` bean exists, so [destroy] never cancels an application's scope. */ + internal var ownedScope: CoroutineScope? = null + private set + + /** + * Runs the fire-and-forget cache writes of every ReqShield this aspect creates: the application's bean named + * `reqShieldCoroutineScope` when there is one, otherwise a scope owned by this aspect. + * + * The library registers no bean of its own, so an application bean of that name replaces the default instead of + * clashing with it. Resolved in [setBeanFactory], at startup, so that a bean of that name which is not a + * `CoroutineScope` fails the context refresh instead of being ignored. + */ + private lateinit var scope: CoroutineScope + + /** + * Spring 6.1+ drops the trailing Continuation of a suspend function itself, in both SimpleKeyGenerator + * and MethodBasedEvaluationContext, so the aspect must pass the raw arguments there and filter them itself + * only on older versions. Detected by a class added in 6.1 instead of SpringVersion, which is null when the + * jar manifest lacks Implementation-Version (e.g. shaded jars): guessing wrong there would drop the real + * last argument from the cache key and make different calls share one entry. + */ + private val springDropsContinuationArgument = + ClassUtils.isPresent("org.springframework.aop.framework.CoroutinesUtils", ReqShieldAspect::class.java.classLoader) + private val spelParser = SpelExpressionParser() private val parameterNameDiscoverer = DefaultParameterNameDiscoverer() private var defaultKeyGenerator = SingletonSupplier.of { SimpleKeyGenerator() } @@ -166,7 +196,7 @@ open class ReqShieldAspect( ): String { val method = getTargetMethod(joinPoint) val args = - if (isCoroutineSupportedSpringVersion()) { + if (springDropsContinuationArgument) { joinPoint.args } else { joinPoint.args.filter { it !is Continuation<*> }.toTypedArray() @@ -268,16 +298,41 @@ open class ReqShieldAspect( } } - private fun isCoroutineSupportedSpringVersion(): Boolean { - val version = springVersion ?: return false - val parts = version.split(".") - val major = parts.getOrNull(0)?.toIntOrNull() ?: return false - val minor = parts.getOrNull(1)?.toIntOrNull() ?: return false + override fun setBeanFactory(beanFactory: BeanFactory) { + this.beanFactory = beanFactory + scope = + if (beanFactory.containsBean(SCOPE_BEAN_NAME)) { + // Throws BeanNotOfRequiredTypeException for a bean of another type + beanFactory.getBean(SCOPE_BEAN_NAME, CoroutineScope::class.java) + } else { + createOwnedScope().also { ownedScope = it } + } + } - return major > 6 || (major == 6 && minor >= 1) + /** Closing the application context cancels the owned scope and with it every pending cache write. */ + override fun destroy() { + ownedScope?.cancel() } - override fun setBeanFactory(beanFactory: BeanFactory) { - this.beanFactory = beanFactory + /** + * SupervisorJob keeps one failed write from cancelling the others, and the exception handler is the last-resort + * backstop for anything the write path did not already log. The name tells its writes apart from those of the + * core default scope, which also runs on Dispatchers.IO. + */ + private fun createOwnedScope(): CoroutineScope = + CoroutineScope( + SupervisorJob() + Dispatchers.IO + CoroutineName(OWNED_SCOPE_NAME) + + CoroutineExceptionHandler { _, e -> + log.error("[Req-Shield] background task failed", e) + }, + ) + + companion object { + private val log = LoggerFactory.getLogger(ReqShieldAspect::class.java) + + /** Name of the optional application bean that replaces the scope owned by the aspect. */ + internal const val SCOPE_BEAN_NAME = "reqShieldCoroutineScope" + + internal const val OWNED_SCOPE_NAME = "req-shield-aspect" } } diff --git a/core-spring-webflux-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/config/LibAutoConfiguration.kt b/core-spring-webflux-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/config/LibAutoConfiguration.kt index d84cc55..79b0c06 100644 --- a/core-spring-webflux-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/config/LibAutoConfiguration.kt +++ b/core-spring-webflux-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/config/LibAutoConfiguration.kt @@ -19,19 +19,10 @@ package com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.config import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.aspect.ReqShieldAspect import com.linecorp.cse.reqshield.support.config.LocalLockLimit import com.linecorp.cse.reqshield.support.constant.ConfigValues.MAX_LOCK_ENTRIES_PROPERTY -import kotlinx.coroutines.CoroutineExceptionHandler -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.cancel -import org.slf4j.LoggerFactory -import org.springframework.beans.factory.DisposableBean -import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import org.springframework.context.annotation.EnableAspectJAutoProxy import org.springframework.context.annotation.Import import org.springframework.core.env.Environment -import kotlin.coroutines.CoroutineContext @Configuration @EnableAspectJAutoProxy @@ -47,37 +38,4 @@ open class LibAutoConfiguration( // of failing the context refresh the way Environment's own Long conversion would. LocalLockLimit.applyConfiguredValue(environment.getProperty(MAX_LOCK_ENTRIES_PROPERTY)) } - - /** - * Scope shared by every [com.linecorp.cse.reqshield.kotlin.coroutine.ReqShield] the aspect - * creates, used for the fire-and-forget cache writes. - * - * SupervisorJob keeps one failed write from cancelling the others, and the exception handler is - * the last-resort backstop for anything the write path did not already log. - */ - @Bean - open fun reqShieldCoroutineScope(): CoroutineScope = - ReqShieldCoroutineScope( - SupervisorJob() + Dispatchers.IO + - CoroutineExceptionHandler { _, e -> - log.error("[Req-Shield] background task failed", e) - }, - ) - - /** - * [CoroutineScope] has no `cancel` member, so the bean carries its own shutdown hook: closing - * the application context cancels the scope and with it every pending cache write. - */ - private class ReqShieldCoroutineScope( - override val coroutineContext: CoroutineContext, - ) : CoroutineScope, - DisposableBean { - override fun destroy() { - coroutineContext.cancel() - } - } - - companion object { - private val log = LoggerFactory.getLogger(LibAutoConfiguration::class.java) - } } diff --git a/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/InMemoryAsyncCache.kt b/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/InMemoryAsyncCache.kt index 2abc8a6..96b5bca 100644 --- a/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/InMemoryAsyncCache.kt +++ b/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/InMemoryAsyncCache.kt @@ -19,6 +19,8 @@ package com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.aspect import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.cache.AsyncCache import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.cache.GlobalLockSupport import com.linecorp.cse.reqshield.support.model.ReqShieldData +import kotlinx.coroutines.CoroutineName +import kotlinx.coroutines.currentCoroutineContext import java.util.concurrent.ConcurrentHashMap class InMemoryAsyncCache : @@ -31,6 +33,14 @@ class InMemoryAsyncCache : private val store = ConcurrentHashMap>() private val locks = ConcurrentHashMap() + /** Name of the thread that ran the most recent write, to tell which scope performed it. */ + @Volatile + var lastWriterThread: String? = null + + /** [CoroutineName] of the most recent write, for scopes that share a dispatcher. */ + @Volatile + var lastWriterCoroutineName: String? = null + override suspend fun get(key: String): ReqShieldData? { val now = System.currentTimeMillis() return store[key]?.let { e -> if (now <= e.expiresAt) e.data else null } @@ -41,6 +51,8 @@ class InMemoryAsyncCache : value: ReqShieldData, timeToLiveMillis: Long, ): Boolean { + lastWriterThread = Thread.currentThread().name + lastWriterCoroutineName = currentCoroutineContext()[CoroutineName]?.name val expiresAt = System.currentTimeMillis() + timeToLiveMillis store[key] = Entry(value, expiresAt) return true diff --git a/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspectIntegrationTest.kt b/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspectIntegrationTest.kt index af14308..196defb 100644 --- a/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspectIntegrationTest.kt +++ b/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspectIntegrationTest.kt @@ -20,9 +20,13 @@ import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.annotation.Req import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.annotation.ReqShieldCacheable import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.cache.AsyncCache import com.linecorp.cse.reqshield.spring.webflux.kotlin.coroutine.config.LibAutoConfiguration +import com.linecorp.cse.reqshield.support.spring.withNamedBean import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExecutorCoroutineDispatcher import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.asCoroutineDispatcher import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.delay @@ -35,12 +39,16 @@ import org.junit.jupiter.api.Assertions.assertNull import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test import org.junit.jupiter.api.extension.ExtendWith +import org.springframework.beans.factory.BeanCreationException +import org.springframework.beans.factory.BeanNotOfRequiredTypeException import org.springframework.beans.factory.annotation.Autowired +import org.springframework.context.ApplicationContext import org.springframework.context.annotation.AnnotationConfigApplicationContext import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import org.springframework.test.context.ContextConfiguration import org.springframework.test.context.junit.jupiter.SpringExtension +import java.util.concurrent.Executors import java.util.concurrent.atomic.AtomicInteger import kotlin.test.assertFailsWith @@ -54,17 +62,18 @@ class ReqShieldAspectIntegrationTest { private lateinit var asyncCache: AsyncCache @Autowired - private lateinit var reqShieldCoroutineScope: CoroutineScope + private lateinit var applicationContext: ApplicationContext /** The aspect namespaces every key with the cache name of the annotation. */ private fun cacheKeyOf(key: String) = "$CACHE_NAME::$key" private suspend fun awaitCachePut( key: String, + cache: AsyncCache<*> = asyncCache, timeoutMillis: Long = 1_000, ): Boolean = withTimeoutOrNull(timeoutMillis) { - while (asyncCache.get(cacheKeyOf(key)) == null) { + while (cache.get(cacheKeyOf(key)) == null) { delay(5) } true @@ -134,19 +143,79 @@ class ReqShieldAspectIntegrationTest { } @Test - fun shouldWireTheCoroutineScopeBeanAndCancelItOnShutdown() { - // The autowired bean proves LibAutoConfiguration alone provides the scope (no component scan). - assertFalse(reqShieldCoroutineScope.coroutineContext[Job]!!.isCancelled) + fun libraryShouldNotRegisterACoroutineScopeBean() { + // A library bean would clash with an application bean of the same name + assertTrue(applicationContext.getBeansOfType(CoroutineScope::class.java).isEmpty()) + } + + @Test + fun ownedScopeShouldRunCacheWritesAndBeCancelledWhenTheContextCloses() = + runBlocking { + val context = AnnotationConfigApplicationContext(*CONFIGURATIONS) + val key = "owned-scope-${System.nanoTime()}" + context.getBean(TestService::class.java).get(key) + val cache = context.getBean(InMemoryAsyncCache::class.java) + + assertTrue(awaitCachePut(key, cache), "Timed out waiting for cache put for key=$key") + assertEquals(ReqShieldAspect.OWNED_SCOPE_NAME, cache.lastWriterCoroutineName) + + val ownedScope = context.getBean(ReqShieldAspect::class.java).ownedScope!! + assertFalse(ownedScope.coroutineContext[Job]!!.isCancelled) + context.close() + assertTrue(ownedScope.coroutineContext[Job]!!.isCancelled) + } + + @Test + fun coroutineScopeBeanNamedReqShieldCoroutineScopeShouldReplaceTheDefault() { + val userDispatcher = namedDispatcher("user-scope") + val userScope = CoroutineScope(SupervisorJob() + userDispatcher) + withNamedBean("reqShieldCoroutineScope", userScope, *CONFIGURATIONS) { userContext -> + runBlocking { + val key = "user-scope-${System.nanoTime()}" + userContext.getBean(TestService::class.java).get(key) + val userCache = userContext.getBean(InMemoryAsyncCache::class.java) + + assertTrue(awaitCachePut(key, userCache), "Timed out waiting for cache put for key=$key") + // Coroutine debug mode appends " @coroutine#N" to the thread name + assertTrue(userCache.lastWriterThread!!.startsWith("user-scope"), userCache.lastWriterThread) + assertNull(userContext.getBean(ReqShieldAspect::class.java).ownedScope) + } + } + // The scope belongs to the application, so the library must leave it active + assertFalse(userScope.coroutineContext[Job]!!.isCancelled) + userDispatcher.close() + } - val context = AnnotationConfigApplicationContext(LibAutoConfiguration::class.java, TestConfig::class.java) - val scope = context.getBean("reqShieldCoroutineScope", CoroutineScope::class.java) - assertFalse(scope.coroutineContext[Job]!!.isCancelled) + @Test + fun beanNamedReqShieldCoroutineScopeOfAnotherTypeShouldFailTheRefresh() { + val error = + assertFailsWith { + withNamedBean("reqShieldCoroutineScope", "not a scope", *CONFIGURATIONS) { } + } - context.close() + assertTrue(error.mostSpecificCause is BeanNotOfRequiredTypeException, error.toString()) + } - assertTrue(scope.coroutineContext[Job]?.isCancelled == true) + @Test + fun coroutineScopeBeansWithOtherNamesShouldBeIgnored() { + val otherDispatcher = namedDispatcher("other-scope") + withNamedBean("otherCoroutineScope", CoroutineScope(SupervisorJob() + otherDispatcher), *CONFIGURATIONS) { otherContext -> + runBlocking { + val key = "other-scope-${System.nanoTime()}" + otherContext.getBean(TestService::class.java).get(key) + val otherCache = otherContext.getBean(InMemoryAsyncCache::class.java) + + assertTrue(awaitCachePut(key, otherCache), "Timed out waiting for cache put for key=$key") + assertEquals(ReqShieldAspect.OWNED_SCOPE_NAME, otherCache.lastWriterCoroutineName) + } + } + otherDispatcher.close() } + /** Single daemon thread named [name], so a cache write shows which scope ran it. */ + private fun namedDispatcher(name: String): ExecutorCoroutineDispatcher = + Executors.newSingleThreadExecutor { Thread(it, name).apply { isDaemon = true } }.asCoroutineDispatcher() + @Configuration open class TestConfig { @Bean @@ -171,5 +240,7 @@ class ReqShieldAspectIntegrationTest { companion object { const val CACHE_NAME = "it" + + private val CONFIGURATIONS = arrayOf(LibAutoConfiguration::class.java, TestConfig::class.java) } } diff --git a/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspectTest.kt b/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspectTest.kt index 452a4fe..28e2122 100644 --- a/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspectTest.kt +++ b/core-spring-webflux-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/kotlin/coroutine/aspect/ReqShieldAspectTest.kt @@ -63,7 +63,7 @@ class ReqShieldAspectTest : BaseReqShieldModuleSupportTest { private val asyncCache: AsyncCache = InMemoryAsyncCache() private val joinPoint: ProceedingJoinPoint = mockk() private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) - private val reqShieldAspect: ReqShieldAspect = spyk(ReqShieldAspect(asyncCache, scope)) + private val reqShieldAspect: ReqShieldAspect = spyk(ReqShieldAspect(asyncCache)) private val targetObject = spyk(TestBean()) private val argument = mapOf("x" to "paramX", "y" to "paramY") private val mockContinuation = mockk>() @@ -83,6 +83,9 @@ class ReqShieldAspectTest : BaseReqShieldModuleSupportTest { every { mockContinuation.context } returns EmptyCoroutineContext every { joinPoint.args } returns arrayOf(argument, mockContinuation) every { joinPoint.target } returns targetObject + // The aspect resolves the application's scope bean when the bean factory is set + every { beanFactory.containsBean("reqShieldCoroutineScope") } returns true + every { beanFactory.getBean("reqShieldCoroutineScope", CoroutineScope::class.java) } returns scope reqShieldAspect.setBeanFactory(beanFactory) } @@ -266,7 +269,7 @@ class ReqShieldAspectTest : BaseReqShieldModuleSupportTest { fun globalLockRequiresTheCacheToImplementGlobalLockSupport() = runTest { val plainCache = mockk>() - val aspect = spyk(ReqShieldAspect(plainCache, scope)) + val aspect = spyk(ReqShieldAspect(plainCache)) aspect.setBeanFactory(beanFactory) every { aspect.getTargetMethod(joinPoint) } returns methodOf(TestBean::cacheableWithGlobalLock.name) @@ -289,14 +292,13 @@ class ReqShieldAspectTest : BaseReqShieldModuleSupportTest { coEvery { lockableCache.globalLock(any(), capture(tokenSlot), any()) } returns true coEvery { lockableCache.globalUnLock(any(), any()) } returns true - val lockScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) - val aspect = spyk(ReqShieldAspect(lockableCache, lockScope)) + val aspect = spyk(ReqShieldAspect(lockableCache)) aspect.setBeanFactory(beanFactory) every { aspect.getTargetMethod(joinPoint) } returns methodOf(TestBean::cacheableWithGlobalLock.name) coEvery { joinPoint.proceed(any>()) } coAnswers { targetObject.cacheableWithGlobalLock(argument) } aspect.aroundReqShieldCacheable(joinPoint) - lockScope.awaitBackgroundWrites() + scope.awaitBackgroundWrites() val expectedLockKey = "$LOCK_KEY_PREFIX${namespacedSpelKey}_CREATE" assertTrue(tokenSlot.isCaptured, "the aspect never called globalLock") diff --git a/core-spring-webflux/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspect.kt b/core-spring-webflux/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspect.kt index f6448a1..bb6091e 100644 --- a/core-spring-webflux/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspect.kt +++ b/core-spring-webflux/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspect.kt @@ -30,7 +30,6 @@ import org.aspectj.lang.reflect.MethodSignature import org.springframework.aop.support.AopUtils import org.springframework.beans.factory.BeanFactory import org.springframework.beans.factory.BeanFactoryAware -import org.springframework.beans.factory.annotation.Qualifier import org.springframework.cache.interceptor.KeyGenerator import org.springframework.cache.interceptor.SimpleKeyGenerator import org.springframework.context.expression.MethodBasedEvaluationContext @@ -43,15 +42,27 @@ import org.springframework.util.StringUtils import org.springframework.util.function.SingletonSupplier import reactor.core.publisher.Mono import reactor.core.scheduler.Scheduler +import reactor.core.scheduler.Schedulers import java.lang.reflect.Method import java.util.concurrent.ConcurrentHashMap @Aspect open class ReqShieldAspect( private val asyncCache: AsyncCache, - @Qualifier("reqShieldScheduler") private val scheduler: Scheduler, ) : BeanFactoryAware { private lateinit var beanFactory: BeanFactory + + /** + * Runs the asynchronous cache writes and the cache polling of every ReqShield this aspect creates: the + * application's bean named `reqShieldScheduler` when there is one, otherwise Reactor's process-wide + * [Schedulers.boundedElastic], which is shared and therefore never disposed here. + * + * The library registers no bean of its own, so an application bean of that name replaces the default instead of + * clashing with it. Resolved in [setBeanFactory], at startup, so that a bean of that name which is not a + * `Scheduler` fails the context refresh instead of being ignored. + */ + private lateinit var scheduler: Scheduler + private val spelParser = SpelExpressionParser() private val parameterNameDiscoverer = DefaultParameterNameDiscoverer() private val defaultKeyGenerator = SingletonSupplier.of { SimpleKeyGenerator() } @@ -251,5 +262,17 @@ open class ReqShieldAspect( override fun setBeanFactory(beanFactory: BeanFactory) { this.beanFactory = beanFactory + scheduler = + if (beanFactory.containsBean(SCHEDULER_BEAN_NAME)) { + // Throws BeanNotOfRequiredTypeException for a bean of another type + beanFactory.getBean(SCHEDULER_BEAN_NAME, Scheduler::class.java) + } else { + Schedulers.boundedElastic() + } + } + + companion object { + /** Name of the optional application bean that replaces the shared boundedElastic scheduler. */ + internal const val SCHEDULER_BEAN_NAME = "reqShieldScheduler" } } diff --git a/core-spring-webflux/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/config/LibAutoConfiguration.kt b/core-spring-webflux/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/config/LibAutoConfiguration.kt index 6978632..830b4dd 100644 --- a/core-spring-webflux/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/config/LibAutoConfiguration.kt +++ b/core-spring-webflux/src/main/kotlin/com/linecorp/cse/reqshield/spring/webflux/config/LibAutoConfiguration.kt @@ -19,13 +19,10 @@ package com.linecorp.cse.reqshield.spring.webflux.config import com.linecorp.cse.reqshield.spring.webflux.aspect.ReqShieldAspect import com.linecorp.cse.reqshield.support.config.LocalLockLimit import com.linecorp.cse.reqshield.support.constant.ConfigValues.MAX_LOCK_ENTRIES_PROPERTY -import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import org.springframework.context.annotation.EnableAspectJAutoProxy import org.springframework.context.annotation.Import import org.springframework.core.env.Environment -import reactor.core.scheduler.Scheduler -import reactor.core.scheduler.Schedulers @Configuration @EnableAspectJAutoProxy @@ -41,13 +38,4 @@ open class LibAutoConfiguration( // of failing the context refresh the way Environment's own Long conversion would. LocalLockLimit.applyConfiguredValue(environment.getProperty(MAX_LOCK_ENTRIES_PROPERTY)) } - - /** - * Scheduler shared by every [com.linecorp.cse.reqshield.reactor.ReqShield] the aspect creates, used for - * the asynchronous cache writes and for polling the cache while another request holds the lock. - * - * [Schedulers.boundedElastic] is Reactor's process-wide instance, so the container must never dispose it. - */ - @Bean(destroyMethod = "") - open fun reqShieldScheduler(): Scheduler = Schedulers.boundedElastic() } diff --git a/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/InMemoryAsyncCache.kt b/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/InMemoryAsyncCache.kt index ffaa7f9..0e6d95f 100644 --- a/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/InMemoryAsyncCache.kt +++ b/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/InMemoryAsyncCache.kt @@ -36,6 +36,10 @@ class InMemoryAsyncCache : private val store = ConcurrentHashMap>() private val locks = ConcurrentHashMap() + /** Name of the thread that ran the most recent write, to tell which scheduler performed it. */ + @Volatile + var lastWriterThread: String? = null + override fun get(key: String): Mono?> = Mono.fromCallable { val now = System.currentTimeMillis() @@ -48,6 +52,7 @@ class InMemoryAsyncCache : timeToLiveMillis: Long, ): Mono = Mono.fromCallable { + lastWriterThread = Thread.currentThread().name val expiresAt = System.currentTimeMillis() + timeToLiveMillis store[key] = Entry(value, expiresAt) true diff --git a/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspectIntegrationTest.kt b/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspectIntegrationTest.kt index 2d9b5b0..dfeca17 100644 --- a/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspectIntegrationTest.kt +++ b/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspectIntegrationTest.kt @@ -21,22 +21,30 @@ import com.linecorp.cse.reqshield.spring.webflux.annotation.ReqShieldCacheable import com.linecorp.cse.reqshield.spring.webflux.cache.AsyncCache import com.linecorp.cse.reqshield.spring.webflux.config.LibAutoConfiguration import com.linecorp.cse.reqshield.support.model.ReqShieldData +import com.linecorp.cse.reqshield.support.spring.withNamedBean +import org.awaitility.Awaitility.await import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse import org.junit.jupiter.api.Assertions.assertNotNull import org.junit.jupiter.api.Assertions.assertNull import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertThrows import org.junit.jupiter.api.extension.ExtendWith +import org.springframework.beans.factory.BeanCreationException +import org.springframework.beans.factory.BeanNotOfRequiredTypeException import org.springframework.beans.factory.annotation.Autowired +import org.springframework.context.ApplicationContext import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import org.springframework.test.context.ContextConfiguration import org.springframework.test.context.junit.jupiter.SpringExtension import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.core.scheduler.Scheduler import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier +import java.time.Duration import java.util.concurrent.atomic.AtomicInteger private const val INTEGRATION_CACHE_NAME = "it" @@ -50,6 +58,9 @@ class ReqShieldAspectIntegrationTest { @Autowired private lateinit var asyncCache: AsyncCache + @Autowired + private lateinit var applicationContext: ApplicationContext + private fun cacheKey(key: String) = "$INTEGRATION_CACHE_NAME::$key" /** ReqShield writes the cache asynchronously, so a test that needs the entry has to wait for it. */ @@ -138,6 +149,61 @@ class ReqShieldAspectIntegrationTest { assertEquals(1, service.getGlobalLockCount(), "The backend should be called once") } + @Test + fun libraryShouldNotRegisterASchedulerBean() { + // A library bean would clash with an application bean of the same name + assertTrue(applicationContext.getBeansOfType(Scheduler::class.java).isEmpty()) + } + + @Test + fun cacheWritesShouldRunOnTheSharedBoundedElasticSchedulerByDefault() { + val key = "default-scheduler-${System.nanoTime()}" + service.get(key).block() + + assertTrue(awaitCachePut(cacheKey(key))) + assertTrue((asyncCache as InMemoryAsyncCache).lastWriterThread!!.startsWith("boundedElastic-")) + } + + @Test + fun schedulerBeanNamedReqShieldSchedulerShouldReplaceTheDefault() { + val userScheduler = Schedulers.newSingle("user-scheduler", true) + withNamedBean("reqShieldScheduler", userScheduler, *CONFIGURATIONS) { userContext -> + val key = "user-scheduler-${System.nanoTime()}" + userContext.getBean(TestService::class.java).get(key).block() + val userCache = userContext.getBean(InMemoryAsyncCache::class.java) + + await().atMost(Duration.ofSeconds(5)).until { userCache.get(cacheKey(key)).block() != null } + assertTrue(userCache.lastWriterThread!!.startsWith("user-scheduler")) + } + // The scheduler belongs to the application, so the library must leave it running + assertFalse(userScheduler.isDisposed) + userScheduler.dispose() + } + + @Test + fun beanNamedReqShieldSchedulerOfAnotherTypeShouldFailTheRefresh() { + val error = + assertThrows { + withNamedBean("reqShieldScheduler", "not a scheduler", *CONFIGURATIONS) { } + } + + assertTrue(error.mostSpecificCause is BeanNotOfRequiredTypeException, error.toString()) + } + + @Test + fun schedulerBeansWithOtherNamesShouldBeIgnored() { + val otherScheduler = Schedulers.newSingle("other-scheduler", true) + withNamedBean("otherScheduler", otherScheduler, *CONFIGURATIONS) { otherContext -> + val key = "other-scheduler-${System.nanoTime()}" + otherContext.getBean(TestService::class.java).get(key).block() + val otherCache = otherContext.getBean(InMemoryAsyncCache::class.java) + + await().atMost(Duration.ofSeconds(5)).until { otherCache.get(cacheKey(key)).block() != null } + assertTrue(otherCache.lastWriterThread!!.startsWith("boundedElastic-")) + } + otherScheduler.dispose() + } + @Configuration open class TestConfig { @Bean @@ -172,6 +238,10 @@ class ReqShieldAspectIntegrationTest { @ReqShieldCacheEvict(cacheName = INTEGRATION_CACHE_NAME, key = "#key") open fun evictFailing(key: String): Mono = Mono.error(IllegalStateException("eviction must not happen")) } + + companion object { + private val CONFIGURATIONS = arrayOf(LibAutoConfiguration::class.java, TestConfig::class.java) + } } /** diff --git a/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspectTest.kt b/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspectTest.kt index f6103e1..045c540 100644 --- a/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspectTest.kt +++ b/core-spring-webflux/src/test/kotlin/com/linecorp/cse/reqshield/spring/webflux/aspect/ReqShieldAspectTest.kt @@ -46,7 +46,7 @@ import kotlin.test.assertTrue class ReqShieldAspectTest : BaseReqShieldModuleSupportTest { private val asyncCache: AsyncCache = InMemoryAsyncCache() private val joinPoint = mockk() - private val reqShieldAspect = spyk(ReqShieldAspect(asyncCache, Schedulers.boundedElastic())) + private val reqShieldAspect = spyk(ReqShieldAspect(asyncCache)) private val targetObject = spyk(TestBean()) private val argument = mapOf("x" to "paramX", "y" to "paramY") @@ -66,6 +66,8 @@ class ReqShieldAspectTest : BaseReqShieldModuleSupportTest { fun setUp() { every { joinPoint.args } returns arrayOf(argument) every { joinPoint.target } returns targetObject + // No application scheduler bean, so the aspect falls back to the shared boundedElastic + every { beanFactory.containsBean("reqShieldScheduler") } returns false reqShieldAspect.setBeanFactory(beanFactory) } @@ -302,7 +304,7 @@ class ReqShieldAspectTest : BaseReqShieldModuleSupportTest { @Test fun globalLockRequiresTheCacheToImplementGlobalLockSupport() { - val localOnlyAspect = spyk(ReqShieldAspect(LocalOnlyAsyncCache(), Schedulers.boundedElastic())) + val localOnlyAspect = spyk(ReqShieldAspect(LocalOnlyAsyncCache())) localOnlyAspect.setBeanFactory(beanFactory) every { localOnlyAspect.getTargetMethod(joinPoint) } returns findTestBeanMethod(TestBean::cacheableWithGlobalLock.name) diff --git a/core-spring/src/main/kotlin/com/linecorp/cse/reqshield/spring/aspect/ReqShieldAspect.kt b/core-spring/src/main/kotlin/com/linecorp/cse/reqshield/spring/aspect/ReqShieldAspect.kt index 2c83e55..7b228c1 100644 --- a/core-spring/src/main/kotlin/com/linecorp/cse/reqshield/spring/aspect/ReqShieldAspect.kt +++ b/core-spring/src/main/kotlin/com/linecorp/cse/reqshield/spring/aspect/ReqShieldAspect.kt @@ -29,7 +29,7 @@ import org.aspectj.lang.reflect.MethodSignature import org.springframework.aop.support.AopUtils import org.springframework.beans.factory.BeanFactory import org.springframework.beans.factory.BeanFactoryAware -import org.springframework.beans.factory.annotation.Qualifier +import org.springframework.beans.factory.DisposableBean import org.springframework.cache.interceptor.KeyGenerator import org.springframework.cache.interceptor.SimpleKeyGenerator import org.springframework.context.expression.MethodBasedEvaluationContext @@ -43,13 +43,33 @@ import org.springframework.util.function.SingletonSupplier import java.lang.reflect.Method import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.Executor +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors +import java.util.concurrent.atomic.AtomicLong @Aspect class ReqShieldAspect( private val reqShieldCache: ReqShieldCache, - @Qualifier("reqShieldExecutor") private val executor: Executor, -) : BeanFactoryAware { +) : BeanFactoryAware, + DisposableBean { private lateinit var beanFactory: BeanFactory + + /** Set only when no `reqShieldExecutor` bean exists, so [destroy] never shuts down an application's pool. */ + internal var ownedExecutor: ExecutorService? = null + private set + + /** + * Runs the asynchronous cache writes of every ReqShield this aspect creates: the application's bean named + * `reqShieldExecutor` when there is one, otherwise a pool owned by this aspect. + * + * The library deliberately registers no `Executor` bean of its own. Spring Boot backs off its + * `applicationTaskExecutor` when another `Executor` bean exists, so a library bean could silently change + * the executor behind `@Async` in applications that merely have req-shield on the classpath. + * Resolved in [setBeanFactory], at startup, so that a bean of that name which is not an `Executor` fails the + * context refresh instead of being ignored. + */ + private lateinit var executor: Executor + private val spelParser = SpelExpressionParser() private val parameterNameDiscoverer = DefaultParameterNameDiscoverer() private val defaultKeyGenerator = SingletonSupplier.of { SimpleKeyGenerator() } @@ -230,5 +250,35 @@ class ReqShieldAspect( override fun setBeanFactory(beanFactory: BeanFactory) { this.beanFactory = beanFactory + executor = + if (beanFactory.containsBean(EXECUTOR_BEAN_NAME)) { + // Throws BeanNotOfRequiredTypeException for a bean of another type + beanFactory.getBean(EXECUTOR_BEAN_NAME, Executor::class.java) + } else { + createOwnedExecutor().also { ownedExecutor = it } + } + } + + override fun destroy() { + ownedExecutor?.shutdown() + } + + private fun createOwnedExecutor(): ExecutorService { + val threadCounter = AtomicLong(0) + + // Daemon threads, so a pending cache write can never block JVM shutdown. + // Named apart from the core default pool so a thread dump shows which one ran a write. + return Executors.newScheduledThreadPool( + maxOf(2, Runtime.getRuntime().availableProcessors() * 2), + ) { runnable -> + Thread(runnable, "req-shield-aspect-executor-${threadCounter.incrementAndGet()}").apply { + isDaemon = true + } + } + } + + companion object { + /** Name of the optional application bean that replaces the pool owned by the aspect. */ + internal const val EXECUTOR_BEAN_NAME = "reqShieldExecutor" } } diff --git a/core-spring/src/main/kotlin/com/linecorp/cse/reqshield/spring/config/LibAutoConfiguration.kt b/core-spring/src/main/kotlin/com/linecorp/cse/reqshield/spring/config/LibAutoConfiguration.kt index 27bb656..dd15f57 100644 --- a/core-spring/src/main/kotlin/com/linecorp/cse/reqshield/spring/config/LibAutoConfiguration.kt +++ b/core-spring/src/main/kotlin/com/linecorp/cse/reqshield/spring/config/LibAutoConfiguration.kt @@ -19,14 +19,10 @@ package com.linecorp.cse.reqshield.spring.config import com.linecorp.cse.reqshield.spring.aspect.ReqShieldAspect import com.linecorp.cse.reqshield.support.config.LocalLockLimit import com.linecorp.cse.reqshield.support.constant.ConfigValues.MAX_LOCK_ENTRIES_PROPERTY -import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import org.springframework.context.annotation.EnableAspectJAutoProxy import org.springframework.context.annotation.Import import org.springframework.core.env.Environment -import java.util.concurrent.ExecutorService -import java.util.concurrent.Executors -import java.util.concurrent.atomic.AtomicLong @Configuration @EnableAspectJAutoProxy @@ -42,25 +38,4 @@ open class LibAutoConfiguration( // of failing the context refresh the way Environment's own Long conversion would. LocalLockLimit.applyConfiguredValue(environment.getProperty(MAX_LOCK_ENTRIES_PROPERTY)) } - - /** - * Pool shared by every [com.linecorp.cse.reqshield.ReqShield] the aspect creates, used for the - * asynchronous cache writes. - * - * Declared as an [ExecutorService] because this pool is owned by the context: Spring's inferred - * destroy method calls [ExecutorService.shutdown] when the context is closed. The threads are - * daemons anyway so a pending task can never block JVM shutdown. - */ - @Bean - open fun reqShieldExecutor(): ExecutorService { - val threadCounter = AtomicLong(0) - - return Executors.newScheduledThreadPool( - maxOf(2, Runtime.getRuntime().availableProcessors() * 2), - ) { runnable -> - Thread(runnable, "req-shield-executor-${threadCounter.incrementAndGet()}").apply { - isDaemon = true - } - } - } } diff --git a/core-spring/src/test/kotlin/aspect/ReqShieldAspectIntegrationTest.kt b/core-spring/src/test/kotlin/aspect/ReqShieldAspectIntegrationTest.kt index ab388fc..29e5ee1 100644 --- a/core-spring/src/test/kotlin/aspect/ReqShieldAspectIntegrationTest.kt +++ b/core-spring/src/test/kotlin/aspect/ReqShieldAspectIntegrationTest.kt @@ -18,9 +18,11 @@ package aspect import com.linecorp.cse.reqshield.spring.annotation.ReqShieldCacheEvict import com.linecorp.cse.reqshield.spring.annotation.ReqShieldCacheable +import com.linecorp.cse.reqshield.spring.aspect.ReqShieldAspect import com.linecorp.cse.reqshield.spring.cache.ReqShieldCache import com.linecorp.cse.reqshield.spring.config.LibAutoConfiguration import com.linecorp.cse.reqshield.support.model.ReqShieldData +import com.linecorp.cse.reqshield.support.spring.withNamedBean import org.awaitility.Awaitility.await import org.junit.jupiter.api.AfterEach import org.junit.jupiter.api.Assertions.assertEquals @@ -29,20 +31,23 @@ import org.junit.jupiter.api.Assertions.assertNull import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows +import org.springframework.beans.factory.BeanCreationException +import org.springframework.beans.factory.BeanNotOfRequiredTypeException import org.springframework.context.annotation.AnnotationConfigApplicationContext import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import java.time.Duration import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.CountDownLatch -import java.util.concurrent.ExecutorService +import java.util.concurrent.Executor import java.util.concurrent.Executors import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger /** * Wires the aspect the way a consumer does - through [LibAutoConfiguration] only - so it also proves - * that the `@Import` of the aspect and the `reqShieldExecutor` bean work without component scanning. + * that the `@Import` of the aspect works without component scanning. * * spring-test is not a test dependency of this module, so the context is driven directly. */ @@ -53,7 +58,7 @@ class ReqShieldAspectIntegrationTest { @BeforeEach fun setUp() { - context = AnnotationConfigApplicationContext(LibAutoConfiguration::class.java, TestConfig::class.java) + context = AnnotationConfigApplicationContext(*CONFIGURATIONS) service = context.getBean(TestService::class.java) cache = context.getBean(InMemoryReqShieldCache::class.java) } @@ -64,8 +69,62 @@ class ReqShieldAspectIntegrationTest { } @Test - fun executorBeanShouldBeProvidedByTheAutoConfiguration() { - assertNotNull(context.getBean("reqShieldExecutor", ExecutorService::class.java)) + fun libraryShouldNotRegisterAnyExecutorBean() { + // Spring Boot can back off its applicationTaskExecutor when another Executor bean exists + assertTrue(context.getBeansOfType(Executor::class.java).isEmpty()) + } + + @Test + fun cacheWritesShouldRunOnThePoolOwnedByTheAspectAndStopWithTheContext() { + service.get("owned-pool") + + await().atMost(Duration.ofSeconds(5)).until { cache.get("integration::owned-pool") != null } + assertTrue(cache.lastWriterThread!!.startsWith(OWNED_POOL_THREAD_PREFIX), cache.lastWriterThread) + + val ownedExecutor = context.getBean(ReqShieldAspect::class.java).ownedExecutor + assertNotNull(ownedExecutor) + context.close() + assertTrue(ownedExecutor!!.isShutdown) + } + + @Test + fun executorBeanNamedReqShieldExecutorShouldReplaceTheOwnedPool() { + val userPool = Executors.newSingleThreadExecutor { Thread(it, "user-pool") } + // A plain Executor view, so closing the context does not shut the pool down on its own + val userExecutor = Executor { userPool.execute(it) } + withNamedBean("reqShieldExecutor", userExecutor, *CONFIGURATIONS) { userContext -> + userContext.getBean(TestService::class.java).get("user-pool") + val userCache = userContext.getBean(InMemoryReqShieldCache::class.java) + + await().atMost(Duration.ofSeconds(5)).until { userCache.get("integration::user-pool") != null } + assertEquals("user-pool", userCache.lastWriterThread) + assertNull(userContext.getBean(ReqShieldAspect::class.java).ownedExecutor) + } + // The aspect only ever sees the wrapper here; ReqShieldAspectTest checks that destroy() leaves the pool running + userPool.shutdown() + } + + @Test + fun beanNamedReqShieldExecutorOfAnotherTypeShouldFailTheRefresh() { + val error = + assertThrows { + withNamedBean("reqShieldExecutor", "not an executor", *CONFIGURATIONS) { } + } + + assertTrue(error.mostSpecificCause is BeanNotOfRequiredTypeException, error.toString()) + } + + @Test + fun executorBeansWithOtherNamesShouldBeIgnored() { + val otherPool = Executors.newSingleThreadExecutor { Thread(it, "other-pool") } + withNamedBean("applicationTaskExecutor", Executor { otherPool.execute(it) }, *CONFIGURATIONS) { otherContext -> + otherContext.getBean(TestService::class.java).get("other-pool") + val otherCache = otherContext.getBean(InMemoryReqShieldCache::class.java) + + await().atMost(Duration.ofSeconds(5)).until { otherCache.get("integration::other-pool") != null } + assertTrue(otherCache.lastWriterThread!!.startsWith(OWNED_POOL_THREAD_PREFIX), otherCache.lastWriterThread) + } + otherPool.shutdown() } @Test @@ -135,9 +194,20 @@ class ReqShieldAspectIntegrationTest { } } + companion object { + private val CONFIGURATIONS = arrayOf(LibAutoConfiguration::class.java, TestConfig::class.java) + + /** Threads of the pool owned by the aspect, named apart from the core default pool. */ + private const val OWNED_POOL_THREAD_PREFIX = "req-shield-aspect-executor-" + } + class InMemoryReqShieldCache : ReqShieldCache { private val store = ConcurrentHashMap>() + /** Name of the thread that ran the most recent write, to tell which executor performed it. */ + @Volatile + var lastWriterThread: String? = null + override fun get(key: String): ReqShieldData? = store[key] override fun put( @@ -145,6 +215,7 @@ class ReqShieldAspectIntegrationTest { value: ReqShieldData, timeToLiveMillis: Long, ) { + lastWriterThread = Thread.currentThread().name store[key] = value } diff --git a/core-spring/src/test/kotlin/aspect/ReqShieldAspectTest.kt b/core-spring/src/test/kotlin/aspect/ReqShieldAspectTest.kt index 2fc2ad3..551e139 100644 --- a/core-spring/src/test/kotlin/aspect/ReqShieldAspectTest.kt +++ b/core-spring/src/test/kotlin/aspect/ReqShieldAspectTest.kt @@ -34,7 +34,9 @@ import org.aspectj.lang.ProceedingJoinPoint import org.awaitility.Awaitility import org.junit.jupiter.api.AfterEach import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull import org.junit.jupiter.api.Assertions.assertThrows import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.BeforeEach @@ -47,6 +49,7 @@ import java.lang.reflect.Method import java.time.Duration import java.util.Collections import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.Executor import java.util.concurrent.Executors import java.util.concurrent.TimeUnit @@ -54,9 +57,10 @@ private val log = LoggerFactory.getLogger(ReqShieldAspectTest::class.java) class ReqShieldAspectTest : BaseReqShieldModuleSupportTest { private val executor = Executors.newScheduledThreadPool(2) + private val reqShieldCache: ReqShieldCache = mockk() private val joinPoint = mockk() - private val reqShieldAspect = spyk(ReqShieldAspect(reqShieldCache, executor)) + private val reqShieldAspect = spyk(ReqShieldAspect(reqShieldCache)) private val targetObject = spyk(TestBean()) private val argument = mapOf("x" to "paramX", "y" to "paramY") @@ -73,10 +77,22 @@ class ReqShieldAspectTest : BaseReqShieldModuleSupportTest { fun setUp() { every { joinPoint.target } returns targetObject every { joinPoint.args } returns arrayOf(argument) + // The aspect resolves the application's executor bean when the bean factory is set + every { beanFactory.containsBean("reqShieldExecutor") } returns true + every { beanFactory.getBean("reqShieldExecutor", Executor::class.java) } returns executor reqShieldAspect.setBeanFactory(beanFactory) } + @Test + fun destroyShouldLeaveTheApplicationPoolRunning() { + reqShieldAspect.destroy() + + // The pool belongs to the application, so only the application may shut it down + assertNull(reqShieldAspect.ownedExecutor) + assertFalse(executor.isShutdown) + } + @AfterEach fun tearDown() { executor.shutdownNow() @@ -235,7 +251,7 @@ class ReqShieldAspectTest : BaseReqShieldModuleSupportTest { fun globalLockShouldBeAcquiredAndReleasedWithTheSameToken() { // given val globalLockCache = GlobalLockReqShieldCache() - val aspect = spyk(ReqShieldAspect(globalLockCache, executor)) + val aspect = spyk(ReqShieldAspect(globalLockCache)) aspect.setBeanFactory(beanFactory) every { joinPoint.proceed() } returns methodReturn stubTargetMethod(TestBean::cacheableWithGlobalLock.name, aspect) diff --git a/core/src/main/kotlin/com/linecorp/cse/reqshield/ReqShield.kt b/core/src/main/kotlin/com/linecorp/cse/reqshield/ReqShield.kt index 55f3050..0040c8d 100644 --- a/core/src/main/kotlin/com/linecorp/cse/reqshield/ReqShield.kt +++ b/core/src/main/kotlin/com/linecorp/cse/reqshield/ReqShield.kt @@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory import java.util.concurrent.Callable import java.util.concurrent.CompletableFuture import java.util.concurrent.CompletionException +import java.util.concurrent.RejectedExecutionException private val log = LoggerFactory.getLogger(ReqShield::class.java) @@ -65,15 +66,14 @@ class ReqShield( val token = if (onlyCreateCache) null else reqShieldConfig.keyLock.tryLock(key, lockType) if (onlyCreateCache || token != null) { - CompletableFuture.runAsync({ + runInBackground(key, lockType, token) { val reqShieldData = buildReqShieldData( executeCallable(callable, key, lockType, token), timeToLiveMillis, ) executeSetCacheFunction(reqShieldConfig.setCacheFunction, key, reqShieldData, lockType, token) - }, reqShieldConfig.executor) - .whenComplete { _, e -> if (e != null) logAsyncFailure(key, e) } + } } } @@ -122,14 +122,41 @@ class ReqShield( executeCallable(callable, key, lockType, token), timeToLiveMillis, ) - CompletableFuture.runAsync({ + runInBackground(key, lockType, token) { executeSetCacheFunction(reqShieldConfig.setCacheFunction, key, reqShieldData, lockType, token) - }, reqShieldConfig.executor) - .whenComplete { _, e -> if (e != null) logAsyncFailure(key, e) } + } return reqShieldData } + /** + * Runs [task] on the configured executor. The task releases the lock identified by [token] itself, so when the + * executor rejects it - a saturated bounded pool, or one already shut down - the lock is released here instead. + * Only that background cache write or refresh is dropped: the caller still gets its data. + */ + private fun runInBackground( + key: String, + lockType: LockType, + token: String?, + task: () -> Unit, + ) { + try { + CompletableFuture + .runAsync(task, reqShieldConfig.executor) + .whenComplete { _, e -> if (e != null) logAsyncFailure(key, e) } + } catch (e: RejectedExecutionException) { + log.warn("Executor rejected the background cache task for key '{}', so it is skipped", key, e) + if (token != null) { + // A failed release must not cost the caller its data; the lock then expires on its own + try { + reqShieldConfig.keyLock.unLock(key, lockType, token) + } catch (unlockError: Exception) { + log.error("Failed to unlock key '{}' after the executor rejected its task", key, unlockError) + } + } + } + } + /** * Another request holds the lock: poll the cache until that request publishes its result. * diff --git a/core/src/main/kotlin/com/linecorp/cse/reqshield/config/ReqShieldConfiguration.kt b/core/src/main/kotlin/com/linecorp/cse/reqshield/config/ReqShieldConfiguration.kt index d890ebe..1f91a4b 100644 --- a/core/src/main/kotlin/com/linecorp/cse/reqshield/config/ReqShieldConfiguration.kt +++ b/core/src/main/kotlin/com/linecorp/cse/reqshield/config/ReqShieldConfiguration.kt @@ -49,7 +49,9 @@ data class ReqShieldConfiguration( /** * Executor used for the asynchronous cache writes. Only [Executor.execute] is called, so any * pool works, and the library never shuts the pool down - a caller-supplied one stays the - * caller's to manage. Defaults to a single pool shared by every configuration instance. + * caller's to manage. A task the pool rejects (bounded queue full, or pool shut down) only + * skips that cache write or refresh; the request still gets its data and the lock is released. + * Defaults to a single pool shared by every configuration instance. */ val executor: Executor = sharedExecutor, val decisionForUpdate: Int = DEFAULT_DECISION_FOR_UPDATE, diff --git a/core/src/test/kotlin/com/linecorp/cse/reqshield/ReqShieldTest.kt b/core/src/test/kotlin/com/linecorp/cse/reqshield/ReqShieldTest.kt index bc2a493..00fc73a 100644 --- a/core/src/test/kotlin/com/linecorp/cse/reqshield/ReqShieldTest.kt +++ b/core/src/test/kotlin/com/linecorp/cse/reqshield/ReqShieldTest.kt @@ -20,6 +20,7 @@ import com.linecorp.cse.reqshield.config.ReqShieldConfiguration import com.linecorp.cse.reqshield.config.ReqShieldWorkMode import com.linecorp.cse.reqshield.support.BaseReqShieldTest import com.linecorp.cse.reqshield.support.BaseReqShieldTest.Companion.AWAIT_TIMEOUT +import com.linecorp.cse.reqshield.support.constant.ConfigValues.DEFAULT_LOCK_TIMEOUT_MILLIS import com.linecorp.cse.reqshield.support.constant.ConfigValues.LOCK_KEY_PREFIX import com.linecorp.cse.reqshield.support.constant.ConfigValues.MAX_CONSECUTIVE_GET_CACHE_FAILURES import com.linecorp.cse.reqshield.support.exception.ClientException @@ -41,7 +42,9 @@ import java.lang.reflect.Method import java.time.Duration import java.util.concurrent.Callable import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executor import java.util.concurrent.Executors +import java.util.concurrent.RejectedExecutionException import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicReference @@ -65,6 +68,7 @@ class ReqShieldTest : BaseReqShieldTest { private val value = Product("testId", "testName") private val callable: Callable = mockk() private val createToken = "create-token" + private val rejectingExecutor = Executor { throw RejectedExecutionException("rejected by test") } private val updateToken = "update-token" private var timeToLiveMillis: Long = 10000 @@ -283,6 +287,91 @@ class ReqShieldTest : BaseReqShieldTest { } } + @Test + fun shouldReturnComputedDataAndReleaseLockWhenExecutorRejectsTheCacheWrite() { + every { cacheGetter(key) } returns null + every { keyLock.tryLock(key, LockType.CREATE) } returns createToken + every { keyLock.unLock(key, LockType.CREATE, createToken) } returns true + val shield = ReqShield(ReqShieldConfiguration(cacheSetter, cacheGetter, keyLock = keyLock, executor = rejectingExecutor)) + + val result = shield.getAndSetReqShieldData(key, callable, timeToLiveMillis) + + assertEquals(value, result.value) + verify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, createToken) } + verify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + @Test + fun shouldReturnCachedDataAndReleaseLockWhenExecutorRejectsTheRefresh() { + val cached = updateTargetData(oldValue) + every { cacheGetter(key) } returns cached + every { keyLock.tryLock(key, LockType.UPDATE) } returns updateToken + every { keyLock.unLock(key, LockType.UPDATE, updateToken) } returns true + val shield = ReqShield(ReqShieldConfiguration(cacheSetter, cacheGetter, keyLock = keyLock, executor = rejectingExecutor)) + + assertSame(cached, shield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + verify(exactly = 1) { keyLock.unLock(key, LockType.UPDATE, updateToken) } + verify(exactly = 0) { callable.call() } + } + + @Test + fun shouldReturnComputedDataWhenExecutorRejectsTheCacheWriteAndTheUnlockFails() { + every { cacheGetter(key) } returns null + every { keyLock.tryLock(key, LockType.CREATE) } returns createToken + every { keyLock.unLock(key, LockType.CREATE, createToken) } throws IllegalStateException("lock store down") + val shield = ReqShield(ReqShieldConfiguration(cacheSetter, cacheGetter, keyLock = keyLock, executor = rejectingExecutor)) + + val result = shield.getAndSetReqShieldData(key, callable, timeToLiveMillis) + + assertEquals(value, result.value) + verify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, createToken) } + } + + @Test + fun shouldReturnCachedDataWhenExecutorRejectsTheRefreshAndTheUnlockFails() { + val cached = updateTargetData(oldValue) + every { cacheGetter(key) } returns cached + every { keyLock.tryLock(key, LockType.UPDATE) } returns updateToken + every { keyLock.unLock(key, LockType.UPDATE, updateToken) } throws IllegalStateException("lock store down") + val shield = ReqShield(ReqShieldConfiguration(cacheSetter, cacheGetter, keyLock = keyLock, executor = rejectingExecutor)) + + assertSame(cached, shield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + verify(exactly = 1) { keyLock.unLock(key, LockType.UPDATE, updateToken) } + } + + @Test + fun shouldReturnCachedDataWithoutTouchingLocksWhenExecutorRejectsAnUnlockedRefresh() { + val cached = updateTargetData(oldValue) + every { cacheGetter(key) } returns cached + val shield = + ReqShield( + ReqShieldConfiguration( + cacheSetter, + cacheGetter, + keyLock = keyLock, + executor = rejectingExecutor, + reqShieldWorkMode = ReqShieldWorkMode.ONLY_CREATE_CACHE, + ), + ) + + assertSame(cached, shield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + verify(exactly = 0) { keyLock.tryLock(any(), any()) } + verify(exactly = 0) { keyLock.unLock(any(), any(), any()) } + verify(exactly = 0) { callable.call() } + } + + @Test + fun shouldFreeTheLocalLockRightAwayWhenTheExecutorIsShutDown() { + val isolatedKey = "rejected-write-${java.util.UUID.randomUUID()}" + val shutDownExecutor = Executors.newSingleThreadExecutor().apply { shutdown() } + val shield = ReqShield(ReqShieldConfiguration({ _, _, _ -> true }, { null }, executor = shutDownExecutor)) + + assertEquals(value, shield.getAndSetReqShieldData(isolatedKey, callable, timeToLiveMillis).value) + + // A leaked lock would stay held for lockTimeoutMillis and make the next request wait out its polling budget + assertNotNull(KeyLocalLock(DEFAULT_LOCK_TIMEOUT_MILLIS).tryLock(isolatedKey, LockType.CREATE)) + } + @Test override fun testSetMethodCacheNotExistsAndLocalLockAcquired() { every { cacheGetter.invoke(key) } returns null diff --git a/req-shield-spring-boot3-example/src/test/kotlin/com/linecorp/cse/reqshield/spring3/bootcompat/LibAutoConfigurationBootTest.kt b/req-shield-spring-boot3-example/src/test/kotlin/com/linecorp/cse/reqshield/spring3/bootcompat/LibAutoConfigurationBootTest.kt new file mode 100644 index 0000000..677f0e9 --- /dev/null +++ b/req-shield-spring-boot3-example/src/test/kotlin/com/linecorp/cse/reqshield/spring3/bootcompat/LibAutoConfigurationBootTest.kt @@ -0,0 +1,109 @@ +package com.linecorp.cse.reqshield.spring3.bootcompat + +import com.linecorp.cse.reqshield.spring.annotation.ReqShieldCacheable +import com.linecorp.cse.reqshield.spring.cache.ReqShieldCache +import com.linecorp.cse.reqshield.support.model.ReqShieldData +import org.awaitility.Awaitility.await +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.springframework.boot.SpringBootConfiguration +import org.springframework.boot.WebApplicationType +import org.springframework.boot.autoconfigure.EnableAutoConfiguration +import org.springframework.boot.builder.SpringApplicationBuilder +import org.springframework.context.ConfigurableApplicationContext +import org.springframework.context.annotation.Bean +import java.time.Duration +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors + +/** + * Boots the library the way a Spring Boot application does, through its auto-configuration entry. + * + * Kept outside the example's package so that the example's component scan never picks these configurations up. + */ +class LibAutoConfigurationBootTest { + @Test + fun bootTaskExecutorShouldBeKeptInAnApplicationWithoutSpringMvc() { + // WebMvcAutoConfiguration orders Boot's task executor configuration early; without it, a library + // Executor bean registered first would make Boot back off its applicationTaskExecutor. + run(MinimalApplication::class.java, "spring.autoconfigure.exclude=$WEB_MVC_AUTO_CONFIGURATION").use { + assertTrue(it.containsBean("applicationTaskExecutor")) + } + } + + @Test + fun userDefinedReqShieldExecutorShouldStartAndBeUsed() { + run(ApplicationWithOwnExecutor::class.java).use { + it.getBean(CachedService::class.java).get("boot") + val cache = it.getBean(InMemoryReqShieldCache::class.java) + + await().atMost(Duration.ofSeconds(5)).until { cache.lastWriterThread != null } + assertEquals(USER_EXECUTOR_THREAD, cache.lastWriterThread) + } + } + + private fun run( + source: Class<*>, + vararg properties: String, + ): ConfigurableApplicationContext = + SpringApplicationBuilder(source) + .web(WebApplicationType.NONE) + .properties(*properties) + .run() + + @SpringBootConfiguration + @EnableAutoConfiguration + open class MinimalApplication { + @Bean + open fun reqShieldCache(): ReqShieldCache = InMemoryReqShieldCache() + } + + @SpringBootConfiguration + @EnableAutoConfiguration + open class ApplicationWithOwnExecutor { + @Bean + open fun reqShieldCache(): ReqShieldCache = InMemoryReqShieldCache() + + // Spring shuts the pool down with the context through its inferred destroy method + @Bean + open fun reqShieldExecutor(): ExecutorService = Executors.newSingleThreadExecutor { Thread(it, USER_EXECUTOR_THREAD) } + + @Bean + open fun cachedService(): CachedService = CachedService() + } + + open class CachedService { + @ReqShieldCacheable(cacheName = "boot", timeToLiveMillis = 10_000) + open fun get(key: String): String = "value-$key" + } + + class InMemoryReqShieldCache : ReqShieldCache { + private val store = ConcurrentHashMap>() + + /** Name of the thread that ran the most recent write, to tell which executor performed it. */ + @Volatile + var lastWriterThread: String? = null + + override fun get(key: String): ReqShieldData? = store[key] + + override fun put( + key: String, + value: ReqShieldData, + timeToLiveMillis: Long, + ) { + lastWriterThread = Thread.currentThread().name + store[key] = value + } + + override fun evict(key: String): Boolean = store.remove(key) != null + } + + companion object { + private const val WEB_MVC_AUTO_CONFIGURATION = + "org.springframework.boot.autoconfigure.web.servlet.WebMvcAutoConfiguration" + + private const val USER_EXECUTOR_THREAD = "user-executor" + } +} diff --git a/support/src/main/kotlin/com/linecorp/cse/reqshield/support/utils/TimeUtils.kt b/support/src/main/kotlin/com/linecorp/cse/reqshield/support/utils/TimeUtils.kt index 10a5801..3a00da7 100644 --- a/support/src/main/kotlin/com/linecorp/cse/reqshield/support/utils/TimeUtils.kt +++ b/support/src/main/kotlin/com/linecorp/cse/reqshield/support/utils/TimeUtils.kt @@ -16,7 +16,8 @@ package com.linecorp.cse.reqshield.support.utils -import java.time.LocalDateTime -import java.time.ZoneId - -fun nowToEpochTime() = LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() +/** + * Current time in epoch milliseconds, read straight from the system clock. Going through LocalDateTime and the default + * zone instead would pick the earlier offset during a DST fall-back overlap and run an hour behind for that hour. + */ +fun nowToEpochTime() = System.currentTimeMillis() diff --git a/support/src/testFixtures/kotlin/com/linecorp/cse/reqshield/support/spring/SpringTestSupport.kt b/support/src/testFixtures/kotlin/com/linecorp/cse/reqshield/support/spring/SpringTestSupport.kt new file mode 100644 index 0000000..42f005d --- /dev/null +++ b/support/src/testFixtures/kotlin/com/linecorp/cse/reqshield/support/spring/SpringTestSupport.kt @@ -0,0 +1,39 @@ +/* + * Copyright 2024 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package com.linecorp.cse.reqshield.support.spring + +import org.springframework.context.annotation.AnnotationConfigApplicationContext +import java.util.function.Supplier + +/** + * Runs [block] against a context built from [configurations] plus [bean] registered under [beanName], and closes it. + * + * Bean definition overriding is disallowed as in Spring Boot, so a library bean of the same name fails the refresh. + */ +inline fun withNamedBean( + beanName: String, + bean: T, + vararg configurations: Class<*>, + block: (AnnotationConfigApplicationContext) -> Unit, +) { + val context = AnnotationConfigApplicationContext() + context.setAllowBeanDefinitionOverriding(false) + context.register(*configurations) + context.registerBean(beanName, T::class.java, Supplier { bean }) + context.refresh() + context.use(block) +}