From ee53f7555bc87f881fd13326876097d726778284 Mon Sep 17 00:00:00 2001 From: "kanghyun.yang" Date: Wed, 16 Sep 2026 19:08:20 +0900 Subject: [PATCH] issue #47: Configure Gradle for Maven Central publishing and disable BootJar task --- .github/workflows/publish.yml | 92 +++++++++++ .gitignore | 2 + build.gradle.kts | 75 +++++---- .../reqshield/kotlin/coroutine/ReqShield.kt | 17 ++ .../kotlin/coroutine/ReqShieldTest.kt | 155 ++++++++++++++++++ .../cse/reqshield/reactor/ReqShield.kt | 52 ++++-- .../cse/reqshield/reactor/ReqShieldTest.kt | 141 ++++++++++++++++ .../com/linecorp/cse/reqshield/ReqShield.kt | 15 ++ .../linecorp/cse/reqshield/ReqShieldTest.kt | 140 ++++++++++++++++ gradle.properties | 1 + 10 files changed, 644 insertions(+), 46 deletions(-) create mode 100644 .github/workflows/publish.yml diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..104d27a --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,92 @@ +name: Publish to Maven Central + +on: + workflow_dispatch: + inputs: + target: + description: Publication target + required: true + default: dev + type: choice + options: + - dev + - release + +concurrency: + group: publish-${{ inputs.target }} + cancel-in-progress: true + +jobs: + publish: + name: Publish to Maven Central + runs-on: ubuntu-latest + services: + redis: + image: redis:6.2.7-alpine + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + + steps: + - name: Checkout code + uses: actions/checkout@v5 + + - name: Set up JDK 17 + uses: actions/setup-java@v5 + with: + java-version: '17' + distribution: 'temurin' + cache: gradle + + - name: Validate Gradle wrapper + uses: gradle/actions/wrapper-validation@v4 + + - name: Grant execute permission for gradlew + run: chmod +x gradlew + + - name: Run tests + run: ./gradlew test + env: + TEST_REDIS_HOST: localhost + TEST_REDIS_PORT: 6379 + + - name: Set snapshot version + if: inputs.target == 'dev' + run: sed -i 's/snapshotBuild=false/snapshotBuild=true/' gradle.properties + + - name: Set release version + if: inputs.target == 'release' + run: sed -i 's/snapshotBuild=true/snapshotBuild=false/' gradle.properties + + - name: Build with Gradle + run: ./gradlew build + + - name: Publish Dev Snapshot to Sonatype + if: inputs.target == 'dev' + run: ./gradlew --max-workers=1 publishToSonatype + env: + ORG_GRADLE_PROJECT_sonatypeUsername: ${{ secrets.MAVEN_CENTRAL_USERNAME }} + ORG_GRADLE_PROJECT_sonatypePassword: ${{ secrets.MAVEN_CENTRAL_PASSWORD }} + ORG_GRADLE_PROJECT_signingKeyId: ${{ secrets.GPG_SECRET_KEY_ID }} + ORG_GRADLE_PROJECT_signingKey: ${{ secrets.GPG_SECRET_KEY }} + ORG_GRADLE_PROJECT_signingPassword: ${{ secrets.GPG_SECRET_PASSWORD }} + + - name: Publish Release to Maven Central + if: inputs.target == 'release' + run: ./gradlew --max-workers=1 publishToSonatype closeAndReleaseSonatypeRepository + env: + ORG_GRADLE_PROJECT_sonatypeUsername: ${{ secrets.MAVEN_CENTRAL_USERNAME }} + ORG_GRADLE_PROJECT_sonatypePassword: ${{ secrets.MAVEN_CENTRAL_PASSWORD }} + ORG_GRADLE_PROJECT_signingKeyId: ${{ secrets.GPG_SECRET_KEY_ID }} + ORG_GRADLE_PROJECT_signingKey: ${{ secrets.GPG_SECRET_KEY }} + ORG_GRADLE_PROJECT_signingPassword: ${{ secrets.GPG_SECRET_PASSWORD }} + + - name: Cleanup Gradle Cache + if: always() + run: | + rm -f ~/.gradle/caches/modules-2/modules-2.lock + rm -f ~/.gradle/caches/modules-2/gc.properties diff --git a/.gitignore b/.gitignore index bd11a43..d7d8ec4 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,8 @@ build/ !gradle/wrapper/gradle-wrapper.jar !**/src/main/**/build/ !**/src/test/**/build/ +.codex +.gradle-user-home ### IntelliJ IDEA ### .idea diff --git a/build.gradle.kts b/build.gradle.kts index eb56210..d3581f6 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -24,18 +24,23 @@ plugins { application `maven-publish` `java-library` + signing + id("io.github.gradle-nexus.publish-plugin") version "2.0.0" } +val snapshotBuild = providers.gradleProperty("snapshotBuild").getOrElse("true").toBoolean() + allprojects { group = "com.linecorp.cse.reqshield" - version = "1.0.0" + version = "1.0.0${if (snapshotBuild) "-SNAPSHOT" else ""}" apply { plugin("java-test-fixtures") plugin("maven-publish") plugin("java-library") plugin("jacoco") + plugin("signing") } repositories { @@ -65,7 +70,7 @@ 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-")) { + if (project != rootProject && !project.name.endsWith("-example")) { tasks.withType { dependsOn(tasks.test) violationRules { @@ -100,6 +105,9 @@ subprojects { plugin("org.jlleitschuh.gradle.ktlint") } java { + withJavadocJar() + withSourcesJar() + sourceCompatibility = when (project.name) { in springBoot3ProjectNames -> JavaVersion.VERSION_17 @@ -113,37 +121,9 @@ subprojects { } afterEvaluate { - fun getProfile() = properties["PROFILE"] ?: System.getenv()["PROFILE"] ?: "local" - - fun getVersion() = project.version.toString() - - fun getSemanticPostfix() = - when (getProfile()) { - "real" -> "" - else -> "-SNAPSHOT" - } - - version = "${getVersion()}${getSemanticPostfix()}" + if (project.name.endsWith("-example")) return@afterEvaluate publishing { - repositories { - maven { - fun getUrl(): String = - if (getProfile() == "real") { - "https://oss.sonatype.org/service/local/staging/deploy/maven2/" - } else { - "https://oss.sonatype.org/content/repositories/snapshots/" - } - - url = uri(getUrl()) - - credentials { - username = System.getenv()["NEXUS_USER"] - password = System.getenv()["NEXUS_PASS"] - } - } - } - publications { register("mavenJava", MavenPublication::class) { @@ -154,6 +134,14 @@ subprojects { description.set("LINE Req-Shield") url.set("https://github.com/line/req-shield.git") + developers { + developer { + name.set("LINE Corporation") + organization.set("LY Corporation") + organizationUrl.set("https://www.lycorp.co.jp/en/") + } + } + licenses { license { name.set("The Apache License, Version 2.0") @@ -162,13 +150,32 @@ subprojects { } scm { - url.set("scm:git@github.com:line/req-shield.git") - connection.set("scm:git@github.com:line/req-shield.git") - developerConnection.set("scm:git@github.com:line/req-shield.git") + url.set("https://github.com/line/req-shield") + connection.set("scm:git:https://github.com/line/req-shield.git") + developerConnection.set("scm:git:ssh://git@github.com/line/req-shield.git") } } } } } + + val signingKeyId = providers.gradleProperty("signingKeyId").orNull + val signingKey = providers.gradleProperty("signingKey").orNull + val signingPassword = providers.gradleProperty("signingPassword").orNull + if (signingKey != null) { + signing { + useInMemoryPgpKeys(signingKeyId, signingKey, signingPassword) + sign(publishing.publications["mavenJava"]) + } + } + } +} + +nexusPublishing { + repositories { + sonatype { + nexusUrl.set(uri("https://ossrh-staging-api.central.sonatype.com/service/local/")) + snapshotRepositoryUrl.set(uri("https://central.sonatype.com/repository/maven-snapshots/")) + } } } diff --git a/core-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/ReqShield.kt b/core-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/ReqShield.kt index 4bb625c..fc26b7b 100644 --- a/core-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/ReqShield.kt +++ b/core-kotlin-coroutine/src/main/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/ReqShield.kt @@ -98,6 +98,23 @@ class ReqShield( val onlyUpdateCache = reqShieldConfig.reqShieldWorkMode == ReqShieldWorkMode.ONLY_UPDATE_CACHE val token = if (onlyUpdateCache) null else reqShieldConfig.keyLock.tryLock(key, lockType) + if (token != null) { + var cacheCreationRequired = false + try { + // Another request may have filled the cache between our initial miss and lock acquisition. + val cachedData = executeGetCacheFunction(reqShieldConfig.getCacheFunction, key) + if (cachedData != null) return cachedData + cacheCreationRequired = true + } finally { + // A hit, read failure, or cancellation must release the token; creation owns it on a miss. + if (!cacheCreationRequired) { + withContext(NonCancellable) { + reqShieldConfig.keyLock.unLock(key, lockType, token) + } + } + } + } + return if (onlyUpdateCache || token != null) { createReqShieldData(key, callable, timeToLiveMillis, lockType, token) } else { diff --git a/core-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/ReqShieldTest.kt b/core-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/ReqShieldTest.kt index 9ca0716..0428a83 100644 --- a/core-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/ReqShieldTest.kt +++ b/core-kotlin-coroutine/src/test/kotlin/com/linecorp/cse/reqshield/kotlin/coroutine/ReqShieldTest.kt @@ -38,6 +38,7 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.TimeoutCancellationException +import kotlinx.coroutines.async import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.cancel import kotlinx.coroutines.cancelAndJoin @@ -56,6 +57,7 @@ import org.junit.jupiter.api.assertThrows import java.lang.reflect.InvocationTargetException import java.lang.reflect.Method import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference import kotlin.coroutines.Continuation import kotlin.coroutines.EmptyCoroutineContext import kotlin.test.assertFailsWith @@ -159,6 +161,159 @@ class ReqShieldTest : BaseReqShieldTest { ttl: Long, ): ReqShieldData = cachedData(cachedValue, ttl, createdAt = nowToEpochTime() - (ttl * 0.9).toLong()) + @Test + fun shouldReuseCacheWhenAnEarlierMissResumesAfterAnotherRequestFinishes() = + runTest { + val cached = AtomicReference?>() + var reads = 0 + var calls = 0 + val missObserved = CompletableDeferred() + val resumeMiss = CompletableDeferred() + val isolatedKey = "delayed-miss-${java.util.UUID.randomUUID()}" + val shield = + ReqShield( + ReqShieldConfiguration( + setCacheFunction = { _, data, _ -> + cached.set(data) + true + }, + getCacheFunction = { + if (++reads == 1) { + // Resume this stale miss only after the winner has written the cache and unlocked. + missObserved.complete(Unit) + resumeMiss.await() + null + } else { + cached.get() + } + }, + scope = this@ReqShieldTest.backgroundScope, + ), + ) + val supplier: suspend () -> Product? = { + calls++ + value + } + val delayed = async { shield.getAndSetReqShieldData(isolatedKey, supplier, timeToLiveMillis) } + + missObserved.await() + val winner = shield.getAndSetReqShieldData(isolatedKey, supplier, timeToLiveMillis) + awaitBackgroundWrites() + assertSame(winner, cached.get()) + resumeMiss.complete(Unit) + val result = delayed.await() + awaitBackgroundWrites() + assertEquals(1, calls) + assertSame(winner, result) + } + + @Test + fun shouldReuseNullValuedCacheAfterAcquiringLocalLock() = + runTest { + val cached = cachedData(null, timeToLiveMillis) + coEvery { cacheGetter(key) } returnsMany listOf(null, cached) + coEvery { keyLock.tryLock(key, LockType.CREATE) } returns LOCAL_TOKEN + coEvery { keyLock.unLock(key, LockType.CREATE, LOCAL_TOKEN) } returns true + + assertSame(cached, reqShield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + + coVerify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, LOCAL_TOKEN) } + coVerify(exactly = 0) { callable() } + coVerify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + @Test + fun shouldReuseCacheAfterAcquiringGlobalLockAndReleaseItsToken() = + runTest { + val cached = cachedData(value, timeToLiveMillis) + coEvery { cacheGetter(key) } returnsMany listOf(null, cached) + coEvery { globalLockFunc(any(), any(), any()) } returns true + coEvery { globalUnLockFunc(any(), any()) } returns true + + assertSame(cached, reqShieldForGlobalLock.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + + val owner = slot() + coVerify(exactly = 1) { globalLockFunc(createLockKey, capture(owner), 3000) } + coVerify(exactly = 1) { globalUnLockFunc(createLockKey, owner.captured) } + coVerify(exactly = 0) { callable() } + coVerify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + @Test + fun shouldReleaseLockWhenCacheRecheckFails() = + runTest { + val failure = IllegalStateException("cache recheck failed") + coEvery { cacheGetter(key) } returns null andThenThrows failure + coEvery { keyLock.tryLock(key, LockType.CREATE) } returns LOCAL_TOKEN + coEvery { keyLock.unLock(key, LockType.CREATE, LOCAL_TOKEN) } returns true + + val error = assertFailsWith { reqShield.getAndSetReqShieldData(key, callable, timeToLiveMillis) } + + assertEquals(ErrorCode.GET_CACHE_ERROR, error.errorCode) + assertSame(failure, error.cause) + coVerify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, LOCAL_TOKEN) } + coVerify(exactly = 0) { callable() } + } + + @Test + fun shouldReleaseGlobalLockWhenCacheRecheckIsCancelled() = + runTest { + val recheckStarted = CompletableDeferred() + var reads = 0 + var unlockCompleted = false + coEvery { cacheGetter(key) } coAnswers { + if (++reads == 1) { + null + } else { + recheckStarted.complete(Unit) + awaitCancellation() + } + } + coEvery { globalLockFunc(any(), any(), any()) } returns true + coEvery { globalUnLockFunc(any(), any()) } coAnswers { + // Cleanup must be able to suspend even when the caller has been cancelled. + delay(1) + unlockCompleted = true + true + } + val caller = launch { reqShieldForGlobalLock.getAndSetReqShieldData(key, callable, timeToLiveMillis) } + + recheckStarted.await() + caller.cancelAndJoin() + + assertTrue(caller.isCancelled) + assertTrue(unlockCompleted) + val owner = slot() + coVerify(exactly = 1) { globalLockFunc(createLockKey, capture(owner), 3000) } + coVerify(exactly = 1) { globalUnLockFunc(createLockKey, owner.captured) } + coVerify(exactly = 0) { callable() } + coVerify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + @Test + fun shouldKeepLockUntilAsyncCacheWriteCompletesAfterRecheckMiss() = + runTest { + val writeStarted = CompletableDeferred() + val finishWrite = CompletableDeferred() + coEvery { cacheGetter(key) } returns null + coEvery { cacheSetter(key, any(), any()) } coAnswers { + writeStarted.complete(Unit) + finishWrite.await() + true + } + coEvery { keyLock.tryLock(key, LockType.CREATE) } returns LOCAL_TOKEN + coEvery { keyLock.unLock(key, LockType.CREATE, LOCAL_TOKEN) } returns true + + assertEquals(value, reqShield.getAndSetReqShieldData(key, callable, timeToLiveMillis).value) + writeStarted.await() + coVerify(exactly = 2) { cacheGetter(key) } + coVerify(exactly = 0) { keyLock.unLock(key, LockType.CREATE, LOCAL_TOKEN) } + finishWrite.complete(Unit) + awaitBackgroundWrites() + coVerify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, LOCAL_TOKEN) } + coVerify(exactly = 1) { callable() } + } + @Test override fun testSetMethodCacheNotExistsAndLocalLockAcquired() = runTest { 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 35962ee..bb13e79 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 @@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import java.time.Duration import java.util.concurrent.Callable +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger private val log = LoggerFactory.getLogger(ReqShield::class.java) @@ -134,7 +135,27 @@ class ReqShield( return reqShieldConfig.keyLock .tryLock(key, lockType) - .flatMap { token -> createReqShieldData(key, callable, timeToLiveMillis, lockType, token) } + .flatMap { token -> + val cacheCreationStarted = AtomicBoolean(false) + // Another request may have filled the cache between our initial miss and lock acquisition. + Mono.defer { executeGetCacheFunction(reqShieldConfig.getCacheFunction, key) } + // A global lock can emit on its client's event loop; keep cache reads off that thread. + .subscribeOn(reqShieldConfig.scheduler) + .flatMap { Mono.justOrEmpty(it) } + .switchIfEmpty( + Mono.defer { + val creation = createReqShieldData(key, callable, timeToLiveMillis, lockType, token) + // The existing creation path releases the lock after its asynchronous cache write. + cacheCreationStarted.set(true) + creation + }, + ).doFinally { + // A cache hit, read failure, or cancellation during the recheck must release our token. + if (!cacheCreationStarted.get()) { + releaseLock(key, lockType, token) + } + } + } .switchIfEmpty( Mono.defer { handleLockFailure(key, callable, timeToLiveMillis) @@ -280,20 +301,27 @@ class ReqShield( .doFinally { // Only the holder of a token took a lock, so only it may release one. if (token != null) { - // No retry needed: false means lock already released or expired (not an error) - reqShieldConfig.keyLock - .unLock(key, lockType, token) - .doOnNext { unlocked -> - if (!unlocked) { - log.debug("Lock already released or expired for key '{}'", key) - } - }.subscribe( - { /* success - no action needed */ }, - { e -> log.error("Failed to unlock key '{}': {}", key, e.message, e) }, - ) + releaseLock(key, lockType, token) } }.subscribeOn(reqShieldConfig.scheduler) + private fun releaseLock( + key: String, + lockType: LockType, + token: String, + ) { + // No retry needed: false means lock already released or expired (not an error). + Mono.defer { reqShieldConfig.keyLock.unLock(key, lockType, token) } + .doOnNext { unlocked -> + if (!unlocked) { + log.debug("Lock already released or expired for key '{}'", key) + } + }.subscribe( + { /* success - no action needed */ }, + { e -> log.error("Failed to unlock key '{}': {}", key, e.message, e) }, + ) + } + private fun executeCallable( callable: Callable>, 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 aa79dcc..a49cb03 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 @@ -35,12 +35,15 @@ import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertThrows import reactor.core.publisher.Mono +import reactor.core.publisher.MonoSink +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.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference import kotlin.test.assertNotNull import kotlin.test.assertNull @@ -134,6 +137,144 @@ class ReqShieldTest : BaseReqShieldTest { private fun freshReqShieldData(cachedValue: Product?): ReqShieldData = ReqShieldData(cachedValue, ReqShieldData.Status.NEW, nowToEpochTime(), timeToLiveMillis) + @Test + fun shouldReuseCacheWhenAnEarlierMissResumesAfterAnotherRequestFinishes() { + val cached = AtomicReference?>() + val reads = AtomicInteger() + val calls = AtomicInteger() + lateinit var delayedMiss: MonoSink?> + val shield = + ReqShield( + ReqShieldConfiguration( + setCacheFunction = { _, data, _ -> + Mono.fromCallable { + cached.set(data) + true + } + }, + getCacheFunction = { + Mono.defer { + if (reads.incrementAndGet() == 1) { + // Hold a cache miss until the other request has stored its result and released the lock. + Mono.create { delayedMiss = it } + } else { + Mono.justOrEmpty(cached.get()) + } + } + }, + scheduler = Schedulers.immediate(), + ), + ) + val isolatedKey = "delayed-miss-${java.util.UUID.randomUUID()}" + val supplier = + Callable { + Mono.fromCallable { + calls.incrementAndGet() + value + } + } + + StepVerifier + .create(shield.getAndSetReqShieldData(isolatedKey, supplier, timeToLiveMillis)) + .then { + val winner = shield.getAndSetReqShieldData(isolatedKey, supplier, timeToLiveMillis).block() + assertEquals(value, winner?.value) + assertNotNull(cached.get()) + delayedMiss.success() + }.expectNextMatches { it === cached.get() } + .verifyComplete() + + assertEquals(1, calls.get()) + } + + @Test + fun shouldReuseNullValuedCacheAfterAcquiringLocalLock() { + val cached = freshReqShieldData(null) + every { cacheGetter(key) } returnsMany listOf(Mono.empty(), Mono.just(cached)) + every { keyLock.tryLock(key, LockType.CREATE) } returns Mono.just(token) + every { keyLock.unLock(key, LockType.CREATE, token) } returns Mono.just(true) + + StepVerifier.create(reqShield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + .expectNext(cached) + .verifyComplete() + + verify(timeout = 1000, exactly = 1) { keyLock.unLock(key, LockType.CREATE, token) } + verify(exactly = 0) { callable.call() } + verify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + @Test + fun shouldReuseCacheAfterAcquiringGlobalLockAndReleaseItsToken() { + val cached = freshReqShieldData(value) + every { cacheGetter(key) } returnsMany listOf(Mono.empty(), Mono.just(cached)) + every { globalLockFunc(any(), any(), any()) } returns Mono.just(true) + every { globalUnLockFunc(any(), any()) } returns Mono.just(true) + + StepVerifier.create(reqShieldForGlobalLock.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + .expectNext(cached) + .verifyComplete() + + val owner = slot() + val lockKey = "$LOCK_KEY_PREFIX${key}_${LockType.CREATE.name}" + verify(exactly = 1) { globalLockFunc(lockKey, capture(owner), 3000) } + verify(timeout = 1000, exactly = 1) { globalUnLockFunc(lockKey, owner.captured) } + verify(exactly = 0) { callable.call() } + verify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + @Test + fun shouldReleaseLockWhenCacheRecheckFails() { + val failure = IllegalStateException("cache recheck failed") + every { cacheGetter(key) } returnsMany listOf(Mono.empty(), Mono.error(failure)) + every { keyLock.tryLock(key, LockType.CREATE) } returns Mono.just(token) + every { keyLock.unLock(key, LockType.CREATE, token) } returns Mono.just(true) + + StepVerifier.create(reqShield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + .expectErrorMatches { it is ClientException && it.errorCode == ErrorCode.GET_CACHE_ERROR && it.cause === failure } + .verify() + + verify(timeout = 1000, exactly = 1) { keyLock.unLock(key, LockType.CREATE, token) } + verify(exactly = 0) { callable.call() } + } + + @Test + fun shouldReleaseLockWhenCacheRecheckIsCancelled() { + every { cacheGetter(key) } returnsMany listOf(Mono.empty(), Mono.never()) + every { keyLock.tryLock(key, LockType.CREATE) } returns Mono.just(token) + every { keyLock.unLock(key, LockType.CREATE, token) } returns Mono.just(true) + + StepVerifier.create(reqShield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + .then { verify(timeout = 1000, exactly = 2) { cacheGetter(key) } } + .thenCancel() + .verify() + + verify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, token) } + verify(exactly = 0) { callable.call() } + } + + @Test + fun shouldKeepLockUntilAsyncCacheWriteCompletesAfterRecheckMiss() { + 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 } + .verifyComplete() + + verify(exactly = 2) { cacheGetter(key) } + verify(exactly = 0) { keyLock.unLock(key, LockType.CREATE, token) } + pendingWrite.success(true) + verify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, token) } + verify(exactly = 1) { callable.call() } + } + /** Cached entry that has passed the decisionForUpdate threshold (90% of its TTL). */ private fun updateTargetReqShieldData(cachedValue: Product?): ReqShieldData = ReqShieldData( 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 cdab096..cc94cc9 100644 --- a/core/src/main/kotlin/com/linecorp/cse/reqshield/ReqShield.kt +++ b/core/src/main/kotlin/com/linecorp/cse/reqshield/ReqShield.kt @@ -94,6 +94,21 @@ class ReqShield( // ONLY_UPDATE_CACHE collapses requests on cache update only, so the creation runs without a lock val token = if (onlyUpdateCache) null else reqShieldConfig.keyLock.tryLock(key, lockType) + if (token != null) { + var cacheCreationRequired = false + try { + // Another request may have filled the cache between our initial miss and lock acquisition. + val cachedData = executeGetCacheFunction(reqShieldConfig.getCacheFunction, key) + if (cachedData != null) return cachedData + cacheCreationRequired = true + } finally { + // On a miss, the existing creation path keeps the lock until its asynchronous write finishes. + if (!cacheCreationRequired) { + reqShieldConfig.keyLock.unLock(key, lockType, token) + } + } + } + return if (onlyUpdateCache || token != null) { createReqShieldData(key, callable, timeToLiveMillis, lockType, token) } else { 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 ed3507b..beef766 100644 --- a/core/src/test/kotlin/com/linecorp/cse/reqshield/ReqShieldTest.kt +++ b/core/src/test/kotlin/com/linecorp/cse/reqshield/ReqShieldTest.kt @@ -21,6 +21,7 @@ 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.GET_CACHE_INTERVAL_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 import com.linecorp.cse.reqshield.support.exception.code.ErrorCode @@ -28,6 +29,7 @@ import com.linecorp.cse.reqshield.support.model.Product import com.linecorp.cse.reqshield.support.model.ReqShieldData 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 @@ -43,9 +45,11 @@ import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference import kotlin.test.assertFailsWith import kotlin.test.assertNotNull import kotlin.test.assertNull +import kotlin.test.assertSame class ReqShieldTest : BaseReqShieldTest { private lateinit var reqShield: ReqShield @@ -144,6 +148,142 @@ class ReqShieldTest : BaseReqShieldTest { timeToLiveMillis = timeToLiveMillis, ) + @Test + fun shouldReuseCacheWhenAnEarlierMissResumesAfterAnotherRequestFinishes() { + val cached = AtomicReference?>() + val reads = AtomicInteger() + val calls = AtomicInteger() + val missObserved = CountDownLatch(1) + val resumeMiss = CountDownLatch(1) + val callerExecutor = Executors.newSingleThreadExecutor() + val cacheExecutor = Executors.newSingleThreadScheduledExecutor() + val isolatedKey = "delayed-miss-${java.util.UUID.randomUUID()}" + val shield = + ReqShield( + ReqShieldConfiguration( + setCacheFunction = { _, data, _ -> + cached.set(data) + true + }, + getCacheFunction = { + if (reads.incrementAndGet() == 1) { + // Resume this stale miss only after the winner has written the cache and unlocked. + missObserved.countDown() + check(resumeMiss.await(2, TimeUnit.SECONDS)) + null + } else { + cached.get() + } + }, + executor = cacheExecutor, + ), + ) + val supplier = + Callable { + calls.incrementAndGet() + value + } + + try { + val delayed = + callerExecutor.submit> { + shield.getAndSetReqShieldData(isolatedKey, supplier, timeToLiveMillis) + } + assertTrue(missObserved.await(2, TimeUnit.SECONDS)) + val winner = shield.getAndSetReqShieldData(isolatedKey, supplier, timeToLiveMillis) + // A barrier on the single-thread executor also waits for the cache write's unlock. + cacheExecutor.submit {}.get(2, TimeUnit.SECONDS) + assertSame(winner, cached.get()) + resumeMiss.countDown() + val result = delayed.get(2, TimeUnit.SECONDS) + assertEquals(1, calls.get()) + assertSame(winner, result) + } finally { + resumeMiss.countDown() + callerExecutor.shutdownNow() + cacheExecutor.shutdownNow() + assertTrue(callerExecutor.awaitTermination(2, TimeUnit.SECONDS)) + assertTrue(cacheExecutor.awaitTermination(2, TimeUnit.SECONDS)) + } + } + + @Test + fun shouldReuseNullValuedCacheAfterAcquiringLocalLock() { + val cached = freshData(null) + every { cacheGetter(key) } returnsMany listOf(null, cached) + every { keyLock.tryLock(key, LockType.CREATE) } returns createToken + every { keyLock.unLock(key, LockType.CREATE, createToken) } returns true + + assertSame(cached, reqShield.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + + verify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, createToken) } + verify(exactly = 0) { callable.call() } + verify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + @Test + fun shouldReuseCacheAfterAcquiringGlobalLockAndReleaseItsToken() { + val cached = freshData(value) + every { cacheGetter(key) } returnsMany listOf(null, cached) + every { globalLockFunc(any(), any(), any()) } returns true + every { globalUnLockFunc(any(), any()) } returns true + + assertSame(cached, reqShieldForGlobalLock.getAndSetReqShieldData(key, callable, timeToLiveMillis)) + + val owner = slot() + val lockKey = "$LOCK_KEY_PREFIX${key}_${LockType.CREATE.name}" + verify(exactly = 1) { globalLockFunc(lockKey, capture(owner), 3000) } + verify(exactly = 1) { globalUnLockFunc(lockKey, owner.captured) } + verify(exactly = 0) { callable.call() } + verify(exactly = 0) { cacheSetter(any(), any(), any()) } + } + + @Test + fun shouldReleaseLockWhenCacheRecheckFails() { + val failure = IllegalStateException("cache recheck failed") + every { cacheGetter(key) } returns null andThenThrows failure + every { keyLock.tryLock(key, LockType.CREATE) } returns createToken + every { keyLock.unLock(key, LockType.CREATE, createToken) } returns true + + val error = assertThrows { reqShield.getAndSetReqShieldData(key, callable, timeToLiveMillis) } + + assertEquals(ErrorCode.GET_CACHE_ERROR, error.errorCode) + assertSame(failure, error.cause) + verify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, createToken) } + verify(exactly = 0) { callable.call() } + } + + @Test + fun shouldKeepLockUntilAsyncCacheWriteCompletesAfterRecheckMiss() { + val writeStarted = CountDownLatch(1) + val finishWrite = CountDownLatch(1) + val cacheExecutor = Executors.newSingleThreadScheduledExecutor() + every { cacheGetter(key) } returns null + every { cacheSetter(key, any(), any()) } answers { + writeStarted.countDown() + check(finishWrite.await(2, TimeUnit.SECONDS)) + true + } + 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 = cacheExecutor)) + + try { + assertEquals(value, shield.getAndSetReqShieldData(key, callable, timeToLiveMillis).value) + assertTrue(writeStarted.await(2, TimeUnit.SECONDS)) + verify(exactly = 2) { cacheGetter(key) } + verify(exactly = 0) { keyLock.unLock(key, LockType.CREATE, createToken) } + finishWrite.countDown() + cacheExecutor.submit {}.get(2, TimeUnit.SECONDS) + verify(exactly = 1) { keyLock.unLock(key, LockType.CREATE, createToken) } + verify(exactly = 1) { callable.call() } + } finally { + finishWrite.countDown() + cacheExecutor.shutdownNow() + assertTrue(cacheExecutor.awaitTermination(2, TimeUnit.SECONDS)) + } + } + @Test override fun testSetMethodCacheNotExistsAndLocalLockAcquired() { every { cacheGetter.invoke(key) } returns null diff --git a/gradle.properties b/gradle.properties index 11af114..4e25e17 100644 --- a/gradle.properties +++ b/gradle.properties @@ -15,3 +15,4 @@ # kotlin.code.style=official +snapshotBuild=false