diff --git a/.github/workflows/pull_request_5x.yml b/.github/workflows/pull_request_5x.yml index 5e204668a..905a892b9 100644 --- a/.github/workflows/pull_request_5x.yml +++ b/.github/workflows/pull_request_5x.yml @@ -8,6 +8,16 @@ jobs: build: runs-on: ubuntu-latest + services: + redis: + image: redis:7-alpine + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 10s + --health-timeout 5s + --health-retries 5 steps: - uses: actions/checkout@v4 with: diff --git a/README.md b/README.md index ad5f79f33..01a3e17e7 100644 --- a/README.md +++ b/README.md @@ -128,6 +128,20 @@ public Mono addAccount(@RequestBody Mono account){ ``` +## 本地响应式缓存失效 + +`hsweb-concurrent/hsweb-concurrent-cache` 中的 Guava、Caffeine 缓存在 +`evict`、`evictAll` 处理对应 key 时撤销已登记的加载者，`clear` 撤销处理时已有的加载者。 +加载成功时通过现有加载映射在同一 key 的原子操作内校验身份并同步回填。 +已撤销加载者的完成或错误不会回填缓存、移除之后的新加载者或清除其缓存值。 + +getter 在订阅时才登记加载者并开始缓存读取，避免仅装配 Publisher 就捕获旧值。 +已经开始读取且与 `clear` 重叠的在途请求可以完成原值。`clear` 不是全缓存事务， +不保证撤销并发产生的新订阅，也不改变调用方默认加载来源的新鲜度。 + +Redis 的写入是异步远程操作，仍沿用原失效逻辑；上述本地保证不适用于 +Redis 同节点异步写入或跨节点写入与失效。 + ## License [Apache 2.0](https://github.com/spring-projects/spring-boot/blob/main/LICENSE.txt) diff --git a/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/AbstractReactiveCache.java b/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/AbstractReactiveCache.java index 5bceac178..2cbe8556e 100644 --- a/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/AbstractReactiveCache.java +++ b/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/AbstractReactiveCache.java @@ -16,8 +16,13 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; +/** + * 管理响应式缓存的在途加载与共享结果，存储操作由具体后端实现。 + * 本地同步后端以加载者身份保护回填和失败清理，不提供异步远程后端的原子保证。 + */ @Slf4j public abstract class AbstractReactiveCache implements ReactiveCache { static final Sinks.EmitFailureHandler emitFailureHandler = Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(30)); @@ -58,9 +63,12 @@ private void tryLoad(ContextView context) { source = source .switchIfEmpty((Mono) defaultValue .flatMap(val -> { - return parent.putNow(key, val).thenReturn(val); + return parent.putLoadedValue(this, val).thenReturn(val); })); } + // 在 complete 撤销身份前清理失败，避免旧加载失败误删新加载者或新值。 + source = source.onErrorResume(err -> parent.handleLoaderError( + key, err, parent.evictLoading(key, this))); loading = source.subscribe( val -> { complete(); @@ -95,11 +103,50 @@ private void complete() { public abstract Mono putNow(Object key, Object value); + private Mono putLoadedValue(CacheLoader owner, Object value) { + AtomicReference> write = new AtomicReference<>(Mono.empty()); + cacheLoading.computeIfPresent(owner.key, (key, current) -> { + if (current == owner) { + // 本地缓存同步写入，与同一 key 的失效共用加载映射的原子操作。 + write.set(putNow(key, value)); + } + return current; + }); + return write.get(); + } + + protected final void invalidateLoading(Object key, Runnable invalidation) { + cacheLoading.compute(key, (_key, current) -> { + invalidation.run(); + return null; + }); + } + + protected final void invalidateLoading(Object key, CacheLoader expected, Runnable invalidation) { + cacheLoading.computeIfPresent(key, (_key, current) -> { + if (current == expected) { + invalidation.run(); + return null; + } + return current; + }); + } + + protected final void invalidateAllLoading(Runnable invalidation) { + // 撤销处理时已登记的加载者；与 clear 重叠的新订阅不属于全缓存事务。 + cacheLoading.clear(); + invalidation.run(); + } + + private CacheLoader createLoader(Object key) { + // 读取也延迟到订阅，不能在装配时捕获失效前的本地值。 + return new CacheLoader(this, key, Mono.defer(() -> getNow(key))); + } + @Override @SuppressWarnings("all") public final Mono getMono(Object key) { - return (Mono) cacheLoading - .computeIfAbsent(key, _key -> new CacheLoader(this, _key, getNow(_key))) + return Mono.defer(() -> (Mono) cacheLoading.computeIfAbsent(key, this::createLoader)) .onErrorResume(err -> handleLoaderError(key, err)); } @@ -110,7 +157,7 @@ public final Mono getMono(Object key, Supplier> loader) { return Mono .deferContextual(ctx -> { CacheLoader cacheLoader = cacheLoading.compute(key, (_key, old) -> { - CacheLoader cl = new CacheLoader(this, _key, getNow(_key)); + CacheLoader cl = createLoader(_key); cl.defaultValue(loader.get(), ctx); return cl; }); @@ -122,8 +169,8 @@ public final Mono getMono(Object key, Supplier> loader) { @Override public final Flux getFlux(Object key) { - return (cacheLoading.computeIfAbsent(key, _key -> new CacheLoader(this, _key, getNow(_key)))) - .flatMapIterable(e -> ((List) e)) + return Flux.defer(() -> cacheLoading.computeIfAbsent(key, this::createLoader) + .flatMapIterable(e -> ((List) e))) .onErrorResume(err -> handleLoaderError(key, err)); } @@ -131,7 +178,7 @@ public final Flux getFlux(Object key) { public final Flux getFlux(Object key, Supplier> loader) { return Flux.deferContextual(ctx -> { CacheLoader cacheLoader = cacheLoading.compute(key, (_key, old) -> { - CacheLoader cl = new CacheLoader(this, _key, getNow(_key)); + CacheLoader cl = createLoader(_key); cl.defaultValue(loader.get().collectList(), ctx); return cl; }); @@ -141,9 +188,17 @@ public final Flux getFlux(Object key, Supplier> loader) { } protected Mono handleLoaderError(Object key, Throwable err) { - log.warn("load cache error,key:{},evict it.", key, err); - return evict(key) - .then(Mono.empty()); + return handleLoaderError(key, err, evict(key)); + } + + protected Mono evictLoading(Object key, CacheLoader owner) { + // 异步后端保留原失效逻辑；本地后端在同一原子操作中校验身份和清理。 + return evict(key); + } + + private Mono handleLoaderError(Object key, Throwable err, Mono cleanup) { + log.warn("load cache error,key:{}.", key, err); + return cleanup.then(Mono.empty()); } @Override diff --git a/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/CaffeineReactiveCache.java b/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/CaffeineReactiveCache.java index 84b425213..0313bc204 100644 --- a/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/CaffeineReactiveCache.java +++ b/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/CaffeineReactiveCache.java @@ -10,6 +10,9 @@ import java.util.Arrays; import java.util.Collection; +/** + * Caffeine 的响应式适配，存储操作同步执行，以在途加载者身份保护本地失效。 + */ @SuppressWarnings("all") @AllArgsConstructor public class CaffeineReactiveCache extends AbstractReactiveCache { @@ -18,7 +21,7 @@ public class CaffeineReactiveCache extends AbstractReactiveCache { @Override public Mono evictAll(Iterable key) { - return Mono.fromRunnable(() -> cache.invalidateAll(key)); + return Mono.fromRunnable(() -> key.forEach(k -> invalidateLoading(k, () -> cache.invalidate(k)))); } @Override @@ -46,11 +49,16 @@ public Mono putNow(Object key, Object value) { @Override public Mono evict(Object key) { - return Mono.fromRunnable(() -> cache.invalidate(key)); + return Mono.fromRunnable(() -> invalidateLoading(key, () -> cache.invalidate(key))); + } + + @Override + protected Mono evictLoading(Object key, CacheLoader owner) { + return Mono.fromRunnable(() -> invalidateLoading(key, owner, () -> cache.invalidate(key))); } @Override public Mono clear() { - return Mono.fromRunnable(() -> cache.invalidateAll()); + return Mono.fromRunnable(() -> invalidateAllLoading(cache::invalidateAll)); } } diff --git a/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/GuavaReactiveCache.java b/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/GuavaReactiveCache.java index fa4245283..59237da56 100644 --- a/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/GuavaReactiveCache.java +++ b/hsweb-concurrent/hsweb-concurrent-cache/src/main/java/org/hswebframework/web/cache/supports/GuavaReactiveCache.java @@ -7,6 +7,9 @@ import java.util.Arrays; +/** + * Guava 的响应式适配，存储操作同步执行，以在途加载者身份保护本地失效。 + */ @SuppressWarnings("all") @AllArgsConstructor public class GuavaReactiveCache extends AbstractReactiveCache { @@ -16,7 +19,7 @@ public class GuavaReactiveCache extends AbstractReactiveCache { @Override public Mono evictAll(Iterable key) { - return Mono.fromRunnable(() -> cache.invalidateAll(key)); + return Mono.fromRunnable(() -> key.forEach(k -> invalidateLoading(k, () -> cache.invalidate(k)))); } @Override @@ -32,7 +35,12 @@ public Mono putNow(Object key, Object value) { @Override public Mono evict(Object key) { - return Mono.fromRunnable(() -> cache.invalidate(key)); + return Mono.fromRunnable(() -> invalidateLoading(key, () -> cache.invalidate(key))); + } + + @Override + protected Mono evictLoading(Object key, CacheLoader owner) { + return Mono.fromRunnable(() -> invalidateLoading(key, owner, () -> cache.invalidate(key))); } @Override @@ -51,6 +59,6 @@ public Flux getAll(Object... keys) { @Override public Mono clear() { - return Mono.fromRunnable(() -> cache.invalidateAll()); + return Mono.fromRunnable(() -> invalidateAllLoading(cache::invalidateAll)); } } diff --git a/hsweb-concurrent/hsweb-concurrent-cache/src/test/java/org/hswebframework/web/cache/LocalReactiveCacheInvalidationTest.java b/hsweb-concurrent/hsweb-concurrent-cache/src/test/java/org/hswebframework/web/cache/LocalReactiveCacheInvalidationTest.java new file mode 100644 index 000000000..8a98620e3 --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-cache/src/test/java/org/hswebframework/web/cache/LocalReactiveCacheInvalidationTest.java @@ -0,0 +1,388 @@ +package org.hswebframework.web.cache; + +import com.github.benmanes.caffeine.cache.Caffeine; +import com.google.common.cache.CacheBuilder; +import org.hswebframework.web.cache.supports.CaffeineReactiveCache; +import org.hswebframework.web.cache.supports.GuavaReactiveCache; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; +import reactor.test.StepVerifier; + +import java.time.Duration; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Proxy; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; + +@RunWith(Parameterized.class) +public class LocalReactiveCacheInvalidationTest { + + private static final String KEY = "subject"; + private static final Duration TIMEOUT = Duration.ofSeconds(5); + + @Parameterized.Parameters(name = "{0}") + public static Collection backends() { + return Arrays.asList(new Object[]{"guava"}, new Object[]{"caffeine"}); + } + + private final String backend; + + public LocalReactiveCacheInvalidationTest(String backend) { + this.backend = backend; + } + + @Test + public void evictionPreventsOldLoadFromRepopulatingCache() { + ReactiveCache cache = createCache(); + assertOldLoadCannotRepopulate(cache, cache.evict(KEY)); + } + + @Test + public void batchEvictionPreventsOldLoadFromRepopulatingCache() { + ReactiveCache cache = createCache(); + assertOldLoadCannotRepopulate(cache, cache.evictAll(Collections.singleton(KEY))); + } + + @Test + public void clearPreventsOldLoadFromRepopulatingCache() { + ReactiveCache cache = createCache(); + assertOldLoadCannotRepopulate(cache, cache.clear()); + } + + private static void assertOldLoadCannotRepopulate(ReactiveCache cache, + Mono invalidation) { + Sinks.One pending = Sinks.one(); + AtomicReference oldResult = new AtomicReference<>(); + cache.getMono(KEY, pending::asMono).subscribe(oldResult::set); + invalidation.as(StepVerifier::create).verifyComplete(); + cache.getMono(KEY) + .as(StepVerifier::create) + .expectComplete() + .verify(TIMEOUT); + pending.emitValue("A", Sinks.EmitFailureHandler.FAIL_FAST); + Assert.assertEquals("A", oldResult.get()); + assertFreshValue(cache); + } + + @Test + public void existingSubscribersCanCompleteAfterInvalidation() { + ReactiveCache cache = createCache(); + Sinks.One pending = Sinks.one(); + AtomicReference firstResult = new AtomicReference<>(); + AtomicReference sharedResult = new AtomicReference<>(); + cache.getMono(KEY, pending::asMono).subscribe(firstResult::set); + cache.getMono(KEY).subscribe(sharedResult::set); + cache.evict(KEY).as(StepVerifier::create).verifyComplete(); + pending.emitValue("A", Sinks.EmitFailureHandler.FAIL_FAST); + Assert.assertEquals("A", firstResult.get()); + Assert.assertEquals("A", sharedResult.get()); + assertFreshValue(cache); + } + + @Test + public void oldCompletionAfterClearDoesNotRemoveNewLoader() { + ReactiveCache cache = createCache(); + Sinks.One oldLoad = Sinks.one(); + Sinks.One newLoad = Sinks.one(); + AtomicReference oldResult = new AtomicReference<>(); + AtomicReference newResult = new AtomicReference<>(); + cache.getMono(KEY, oldLoad::asMono).subscribe(oldResult::set); + cache.clear().as(StepVerifier::create).verifyComplete(); + cache.getMono(KEY, newLoad::asMono).subscribe(newResult::set); + + oldLoad.emitValue("A", Sinks.EmitFailureHandler.FAIL_FAST); + Assert.assertEquals("A", oldResult.get()); + newLoad.emitValue("B", Sinks.EmitFailureHandler.FAIL_FAST); + Assert.assertEquals("B", newResult.get()); + cache.getMono(KEY, () -> Mono.error(new AssertionError("new loader was detached"))) + .as(StepVerifier::create) + .expectNext("B") + .verifyComplete(); + } + + @Test + public void currentLoaderErrorAllowsReloadAndHotHit() { + ReactiveCache cache = createCache(); + cache.getMono(KEY, () -> Mono.error(new IllegalStateException("current loader failed"))) + .as(StepVerifier::create) + .expectComplete() + .verify(TIMEOUT); + + AtomicInteger freshLoads = new AtomicInteger(); + Supplier> fresh = () -> Mono.defer(() -> { + freshLoads.incrementAndGet(); + return Mono.just("B"); + }); + cache.getMono(KEY, fresh) + .as(StepVerifier::create) + .expectNext("B") + .verifyComplete(); + cache.getMono(KEY, fresh) + .as(StepVerifier::create) + .expectNext("B") + .verifyComplete(); + Assert.assertEquals(1, freshLoads.get()); + } + + @Test + public void oldErrorDoesNotRemoveNewPendingLoader() { + ReactiveCache cache = createCache(); + Sinks.One oldLoad = Sinks.one(); + Sinks.One newLoad = Sinks.one(); + AtomicInteger freshLoads = new AtomicInteger(); + AtomicInteger oldCompleted = new AtomicInteger(); + AtomicReference oldFailure = new AtomicReference<>(); + AtomicReference newResult = new AtomicReference<>(); + cache.getMono(KEY, oldLoad::asMono) + .subscribe(ignored -> { }, oldFailure::set, oldCompleted::incrementAndGet); + cache.evict(KEY).as(StepVerifier::create).verifyComplete(); + Supplier> fresh = () -> Mono.defer(() -> { + freshLoads.incrementAndGet(); + return newLoad.asMono(); + }); + cache.getMono(KEY, fresh).subscribe(newResult::set); + oldLoad.emitError(new IllegalStateException("old loader failed"), Sinks.EmitFailureHandler.FAIL_FAST); + Assert.assertNull(oldFailure.get()); + Assert.assertEquals(1, oldCompleted.get()); + newLoad.emitValue("B", Sinks.EmitFailureHandler.FAIL_FAST); + Assert.assertEquals("B", newResult.get()); + cache.getMono(KEY, fresh) + .as(StepVerifier::create) + .expectNext("B") + .verifyComplete(); + Assert.assertEquals(1, freshLoads.get()); + } + + @Test + public void oldErrorDoesNotRemoveCompletedNewValue() { + ReactiveCache cache = createCache(); + Sinks.One oldLoad = Sinks.one(); + AtomicInteger freshLoads = new AtomicInteger(); + AtomicInteger oldCompleted = new AtomicInteger(); + AtomicReference oldFailure = new AtomicReference<>(); + cache.getMono(KEY, oldLoad::asMono) + .subscribe(ignored -> { }, oldFailure::set, oldCompleted::incrementAndGet); + cache.evict(KEY).as(StepVerifier::create).verifyComplete(); + Supplier> fresh = () -> Mono.defer(() -> { + freshLoads.incrementAndGet(); + return Mono.just("B"); + }); + cache.getMono(KEY, fresh) + .as(StepVerifier::create) + .expectNext("B") + .verifyComplete(); + oldLoad.emitError(new IllegalStateException("old loader failed"), Sinks.EmitFailureHandler.FAIL_FAST); + Assert.assertNull(oldFailure.get()); + Assert.assertEquals(1, oldCompleted.get()); + cache.getMono(KEY, fresh) + .as(StepVerifier::create) + .expectNext("B") + .verifyComplete(); + Assert.assertEquals(1, freshLoads.get()); + } + + @Test + public void monoAssembledDuringClearDoesNotCaptureOldValue() throws Exception { + assertAssemblingDuringClearDoesNotCaptureOldValue(false); + } + + @Test + public void fluxAssembledDuringClearDoesNotCaptureOldValue() throws Exception { + assertAssemblingDuringClearDoesNotCaptureOldValue(true); + } + + private void assertAssemblingDuringClearDoesNotCaptureOldValue(boolean flux) throws Exception { + CountDownLatch clearWindow = new CountDownLatch(1); + CountDownLatch releaseClear = new CountDownLatch(1); + ReactiveCache cache = createCache(() -> { }, () -> { + clearWindow.countDown(); + awaitRelease(releaseClear); + }); + cache.put(KEY, flux ? Flux.just("A") : Mono.just("A")).block(TIMEOUT); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future clear = executor.submit(() -> cache.clear().block(TIMEOUT)); + Assert.assertTrue(clearWindow.await(5, TimeUnit.SECONDS)); + // 此时加载映射已清理，后端 invalidateAll 尚未执行；仅装配，不订阅。 + Publisher assembled = flux ? cache.getFlux(KEY) : cache.getMono(KEY); + Assert.assertNotNull(assembled); + releaseClear.countDown(); + clear.get(5, TimeUnit.SECONDS); + if (flux) { + cache.getFlux(KEY).as(StepVerifier::create).expectComplete().verify(TIMEOUT); + } else { + cache.getMono(KEY).as(StepVerifier::create).expectComplete().verify(TIMEOUT); + } + } finally { + releaseClear.countDown(); + executor.shutdownNow(); + } + } + + @Test + public void clearPreventsOldFluxLoadFromRepopulatingCache() { + ReactiveCache cache = createCache(); + Sinks.One pending = Sinks.one(); + AtomicReference oldResult = new AtomicReference<>(); + cache.getFlux(KEY, () -> pending.asMono().flux()).subscribe(oldResult::set); + cache.clear().as(StepVerifier::create).verifyComplete(); + pending.emitValue("A", Sinks.EmitFailureHandler.FAIL_FAST); + Assert.assertEquals("A", oldResult.get()); + cache.getFlux(KEY, () -> Flux.just("B")) + .as(StepVerifier::create) + .expectNext("B") + .verifyComplete(); + cache.getFlux(KEY, () -> Flux.error(new AssertionError("new flux was not cached"))) + .as(StepVerifier::create) + .expectNext("B") + .verifyComplete(); + } + + @Test + public void invalidationCannotOvertakeAnAtomicWriteBack() throws Exception { + assertInvalidationCannotOvertakeWriteBack(false); + } + + @Test + public void clearCannotOvertakeAnAtomicWriteBack() throws Exception { + assertInvalidationCannotOvertakeWriteBack(true); + } + + private void assertInvalidationCannotOvertakeWriteBack(boolean clear) throws Exception { + CountDownLatch writeEntered = new CountDownLatch(1); + CountDownLatch releaseWrite = new CountDownLatch(1); + CountDownLatch invalidationStarted = new CountDownLatch(1); + CountDownLatch invalidationCompleted = new CountDownLatch(1); + ReactiveCache cache = createCache(() -> { + writeEntered.countDown(); + try { + if (!releaseWrite.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("write was not released"); + } + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(interrupted); + } + }); + Sinks.One pending = Sinks.one(); + AtomicReference oldResult = new AtomicReference<>(); + cache.getMono(KEY, pending::asMono).subscribe(oldResult::set); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future oldCompletion = executor.submit(() -> + pending.emitValue("A", Sinks.EmitFailureHandler.FAIL_FAST)); + Assert.assertTrue(writeEntered.await(5, TimeUnit.SECONDS)); + Future invalidation = executor.submit(() -> { + invalidationStarted.countDown(); + (clear ? cache.clear() : cache.evict(KEY)).block(TIMEOUT); + invalidationCompleted.countDown(); + }); + Assert.assertTrue(invalidationStarted.await(5, TimeUnit.SECONDS)); + Assert.assertFalse(invalidationCompleted.await(100, TimeUnit.MILLISECONDS)); + releaseWrite.countDown(); + oldCompletion.get(5, TimeUnit.SECONDS); + invalidation.get(5, TimeUnit.SECONDS); + Assert.assertEquals("A", oldResult.get()); + assertFreshValue(cache); + } finally { + releaseWrite.countDown(); + executor.shutdownNow(); + } + } + + private static void assertFreshValue(ReactiveCache cache) { + AtomicInteger freshLoads = new AtomicInteger(); + cache.getMono(KEY, () -> Mono.defer(() -> { + freshLoads.incrementAndGet(); + return Mono.just("B"); + })) + .as(StepVerifier::create) + .expectNext("B") + .verifyComplete(); + Assert.assertEquals(1, freshLoads.get()); + cache.getMono(KEY, () -> Mono.error(new AssertionError("fresh value was not cached"))) + .as(StepVerifier::create) + .expectNext("B") + .verifyComplete(); + } + + private ReactiveCache createCache() { + return createCache(() -> { }); + } + + private ReactiveCache createCache(Runnable beforeOldWrite) { + return createCache(beforeOldWrite, null); + } + + private ReactiveCache createCache(Runnable beforeOldWrite, Runnable beforeClear) { + if ("guava".equals(backend)) { + com.google.common.cache.Cache nativeCache = CacheBuilder.newBuilder().build(); + com.google.common.cache.Cache cache = beforeClear == null + ? nativeCache + : observeClear(com.google.common.cache.Cache.class, nativeCache, beforeClear); + return new GuavaReactiveCache(cache) { + @Override + public Mono putNow(Object key, Object value) { + if ("A".equals(value)) { + beforeOldWrite.run(); + } + return super.putNow(key, value); + } + }; + } + com.github.benmanes.caffeine.cache.Cache nativeCache = Caffeine.newBuilder().build(); + com.github.benmanes.caffeine.cache.Cache cache = beforeClear == null + ? nativeCache + : observeClear(com.github.benmanes.caffeine.cache.Cache.class, nativeCache, beforeClear); + return new CaffeineReactiveCache(cache) { + @Override + public Mono putNow(Object key, Object value) { + if ("A".equals(value)) { + beforeOldWrite.run(); + } + return super.putNow(key, value); + } + }; + } + + @SuppressWarnings("unchecked") + private static T observeClear(Class cacheType, T cache, Runnable beforeClear) { + return (T) Proxy.newProxyInstance(cacheType.getClassLoader(), new Class[]{cacheType}, + (proxy, method, args) -> { + if ("invalidateAll".equals(method.getName()) && method.getParameterCount() == 0) { + beforeClear.run(); + } + try { + return method.invoke(cache, args); + } catch (InvocationTargetException invocation) { + throw invocation.getCause(); + } + }); + } + + private static void awaitRelease(CountDownLatch release) { + try { + if (!release.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("operation was not released"); + } + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(interrupted); + } + } +} diff --git a/hsweb-system/hsweb-system-authorization/README.md b/hsweb-system/hsweb-system-authorization/README.md index ccfd8a6e1..76e9bb19e 100644 --- a/hsweb-system/hsweb-system-authorization/README.md +++ b/hsweb-system/hsweb-system-authorization/README.md @@ -22,3 +22,40 @@ ## 结构 ![uml](./uml.png "uml.png") + +## 统一认证的用户状态 + +`UserEntity` 定义 `STATUS_DISABLED=0`、`STATUS_ENABLED=1`、`STATUS_LOCKED=2`， +及用户维度 options 的状态键 `OPTION_STATUS="status"`。 + +`DefaultReactiveAuthenticationInitializeService#initUserAuthorization` 读取用户， +`doInit` 复用该用户的状态进行准入：禁用或不存在的用户返回空结果； +锁定、状态为空及其他非禁用状态仍可初始化权限并发布 `AuthorizationInitializeEvent`。 +该规则也作用于直接调用 `doInit` 的入口，不依赖事件监听器补充校验。 +真实状态保存在 `SimpleUser` 的 options 中，空状态保留该键及空值， +不写入 `Authentication.attributes`，也不增加用户字段或查询。 + +内部读取权限与对外凭据访问分别准入。默认密码认证仅允许启用用户； +停用和锁定用户不能密码登录。既有用户事件的非启用状态 token 注销规则保持不变。 + +`DefaultReactiveAuthenticationManager#getByUserId` 仅缓存并委派 initializer，不额外查询用户。 +默认 initializer 的非禁用用户冷加载读取一次用户；缓存命中不读取用户。 +无缓存时使用同一 initializer。用户读取异常沿用原链路错误处理，无缓存时向调用方传播。 +状态变更需要在事务提交后发布 `ClearUserAuthorizationCacheEvent`，随后访问按最新状态重新加载。 + +初始化接口、非禁用用户的权限构造和初始化事件语义保持不变。 +该校验不取消已有在途请求。Guava、Caffeine 本地缓存会阻止失效时已撤销的加载者重新回填， +其后续完成或失败也不会清除新的加载者和缓存值。Redis 同节点异步写入及跨节点写入与失效 +不提供同样的原子保证；`clear` 不提供全缓存事务。 + +状态校验仅作用于默认 initializer；自定义 initializer 可以按自身契约从其他来源创建认证， +认证管理器不会要求额外的持久用户。其他认证 Provider 仍按自身契约提供认证， +默认 initializer 返回空不会阻止其他 Provider 返回认证。 +自定义来源为用户委托访问签发凭据时，需要在受信的用户 options 中提供真实状态； +应用认证入口据此判断是否允许访问，不从请求参数或 `Authentication.attributes` 推断状态。 +直接由独立 Token Provider 创建的机器身份不经过该用户加载入口。 + +本轮与应用认证模块同源联合验证 9 类、143 项全部通过，使用当前 API、default 和 cache +源码产物。默认认证测试 6 项确认非禁用用户的实际权限、用户状态和初始化事件、 +冷加载查询一次、热命中零次及密码准入；真实 PostgreSQL、Redis、H2 和 HTTP +验证锁定用户后台读取权限、旧授权码和委托 Token 拒绝、重新启用与删除回滚。 diff --git a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-api/src/main/java/org/hswebframework/web/system/authorization/api/entity/UserEntity.java b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-api/src/main/java/org/hswebframework/web/system/authorization/api/entity/UserEntity.java index 6e0b8f105..379c530dc 100644 --- a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-api/src/main/java/org/hswebframework/web/system/authorization/api/entity/UserEntity.java +++ b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-api/src/main/java/org/hswebframework/web/system/authorization/api/entity/UserEntity.java @@ -36,6 +36,18 @@ @Comment("用户信息") public class UserEntity extends ExtendableEntity implements RecordCreationEntity { + /** 禁用用户，不初始化默认用户权限。 */ + public static final byte STATUS_DISABLED = 0; + + /** 启用用户，允许默认密码认证。 */ + public static final byte STATUS_ENABLED = 1; + + /** 锁定用户，保留权限读取，禁止对外凭据访问。 */ + public static final byte STATUS_LOCKED = 2; + + /** 用户维度 options 中保存真实用户状态的键。 */ + public static final String OPTION_STATUS = "status"; + @Column(length = 128, nullable = false) @NotBlank(message = "姓名不能为空", groups = CreateGroup.class) @Schema(description = "姓名") diff --git a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/DefaultReactiveAuthenticationInitializeService.java b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/DefaultReactiveAuthenticationInitializeService.java index b5f96ac15..dcb9b7ee2 100644 --- a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/DefaultReactiveAuthenticationInitializeService.java +++ b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/DefaultReactiveAuthenticationInitializeService.java @@ -34,6 +34,10 @@ import java.util.function.Function; import java.util.stream.Collectors; +/** + * 初始化默认系统用户的认证，仅排除禁用的持久用户，构造权限并发布认证初始化事件。 + * 用户读取与状态准入在同一链路完成，认证缓存由认证管理器负责。 + */ @Slf4j public class DefaultReactiveAuthenticationInitializeService implements ReactiveAuthenticationInitializeService { @@ -76,29 +80,33 @@ public Mono initUserAuthorization(String userId) { public Mono doInit(Mono userEntityMono) { - return userEntityMono.flatMap(user -> { - SimpleAuthentication authentication = new SimpleAuthentication(); - authentication.setUser(SimpleUser - .builder() - .id(user.getId()) - .name(user.getName()) - .username(user.getUsername()) - .userType(user.getType()) - .build()); + // 权限读取只排除禁用用户；凭据访问另按用户维度中的真实状态准入。 + return userEntityMono + .filter(user -> !Objects.equals(user.getStatus(), UserEntity.STATUS_DISABLED)) + .flatMap(user -> { + SimpleAuthentication authentication = new SimpleAuthentication(); + authentication.setUser(SimpleUser + .builder() + .id(user.getId()) + .name(user.getName()) + .username(user.getUsername()) + .userType(user.getType()) + .options(Collections.singletonMap(UserEntity.OPTION_STATUS, user.getStatus())) + .build()); - return initPermission(authentication) - .defaultIfEmpty(authentication) - .onErrorResume(err -> { - log.warn(err.getMessage(), err); - return Mono.just(authentication); - }) - .flatMap(auth -> { - AuthorizationInitializeEvent event = new AuthorizationInitializeEvent(auth); - return event - .publish(eventPublisher) - .then(Mono.fromSupplier(event::getAuthentication)); - }); - }); + return initPermission(authentication) + .defaultIfEmpty(authentication) + .onErrorResume(err -> { + log.warn(err.getMessage(), err); + return Mono.just(authentication); + }) + .flatMap(auth -> { + AuthorizationInitializeEvent event = new AuthorizationInitializeEvent(auth); + return event + .publish(eventPublisher) + .then(Mono.fromSupplier(event::getAuthentication)); + }); + }); } protected Flux getSettings(List dimensions) { diff --git a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/DefaultReactiveAuthenticationManager.java b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/DefaultReactiveAuthenticationManager.java index 21dd2f350..c95f43ff1 100644 --- a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/DefaultReactiveAuthenticationManager.java +++ b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/DefaultReactiveAuthenticationManager.java @@ -15,6 +15,10 @@ import org.springframework.core.annotation.Order; import reactor.core.publisher.Mono; +/** + * 提供默认用户名密码认证和用户认证缓存。 + * 按用户 ID 获取认证时委派 initializer，认证来源与准入由 initializer 决定。 + */ @Slf4j @Order(100) public class DefaultReactiveAuthenticationManager implements ReactiveAuthenticationManagerProvider { @@ -60,7 +64,7 @@ public Mono authenticate(Mono request) { .filter(PlainTextUsernamePasswordAuthenticationRequest.class::isInstance) .map(PlainTextUsernamePasswordAuthenticationRequest.class::cast) .flatMap(pwdRequest -> reactiveUserService.findByUsernameAndPassword(pwdRequest.getUsername(), pwdRequest.getPassword())) - .filter(user -> Byte.valueOf((byte) 1).equals(user.getStatus())) + .filter(user -> Byte.valueOf(UserEntity.STATUS_ENABLED).equals(user.getStatus())) .map(UserEntity::getId) .flatMap(this::getByUserId); } diff --git a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/RemoveUserTokenWhenUserDisabled.java b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/RemoveUserTokenWhenUserDisabled.java index 80780d087..fe333ff86 100644 --- a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/RemoveUserTokenWhenUserDisabled.java +++ b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/main/java/org/hswebframework/web/system/authorization/defaults/service/RemoveUserTokenWhenUserDisabled.java @@ -3,12 +3,17 @@ import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.hswebframework.web.authorization.token.UserTokenManager; +import org.hswebframework.web.system.authorization.api.entity.UserEntity; import org.hswebframework.web.system.authorization.api.event.UserModifiedEvent; import org.hswebframework.web.system.authorization.api.event.UserStateChangedEvent; import org.springframework.context.event.EventListener; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +/** + * 按既有用户修改和状态变更事件注销非启用用户的 token。 + * 仅处理 token 生命周期，不限制非禁用用户的内部权限初始化。 + */ @AllArgsConstructor @Slf4j public class RemoveUserTokenWhenUserDisabled { @@ -17,7 +22,7 @@ public class RemoveUserTokenWhenUserDisabled { @EventListener public void handleStateChangeEvent(UserModifiedEvent event) { - if (event.getUserEntity().getStatus() != null && event.getUserEntity().getStatus() != 1) { + if (event.getUserEntity().getStatus() != null && event.getUserEntity().getStatus() != UserEntity.STATUS_ENABLED) { event.async( Mono.just(event.getUserEntity().getId()) .flatMap(userTokenManager::signOutByUserId) @@ -27,7 +32,7 @@ public void handleStateChangeEvent(UserModifiedEvent event) { @EventListener public void handleStateChangeEvent(UserStateChangedEvent event) { - if (event.getState() != 1) { + if (event.getState() != UserEntity.STATUS_ENABLED) { event.async( Flux.fromIterable(event.getUserIdList()) .flatMap(userTokenManager::signOutByUserId) diff --git a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/DefaultDimensionUserServiceTest.java b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/DefaultDimensionUserServiceTest.java index 98013d9fd..58271e8a0 100644 --- a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/DefaultDimensionUserServiceTest.java +++ b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/DefaultDimensionUserServiceTest.java @@ -1,7 +1,6 @@ package org.hswebframework.web.system.authorization.defaults.service; import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository; -import org.hswebframework.web.authorization.AuthenticationManager; import org.hswebframework.web.authorization.ReactiveAuthenticationManager; import org.hswebframework.web.system.authorization.api.entity.*; import org.hswebframework.web.system.authorization.api.service.reactive.ReactiveUserService; @@ -14,13 +13,9 @@ import reactor.core.publisher.Mono; import reactor.test.StepVerifier; -import static org.junit.Assert.*; - @RunWith(SpringRunner.class) -@SpringBootTest(classes = { - ReactiveTestApplication.class, - DefaultReactiveUserService.class -}) +@SpringBootTest(classes = ReactiveTestApplication.class, + properties = "spring.r2dbc.generate-unique-name=true") public class DefaultDimensionUserServiceTest { @Autowired @@ -208,4 +203,4 @@ private String initData(String dimensionType, String dimensionId) { return userEntity.getId(); } -} \ No newline at end of file +} diff --git a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/DefaultReactiveAuthenticationManagerTest.java b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/DefaultReactiveAuthenticationManagerTest.java index 302276e9f..5526ecfa6 100644 --- a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/DefaultReactiveAuthenticationManagerTest.java +++ b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/DefaultReactiveAuthenticationManagerTest.java @@ -1,30 +1,47 @@ package org.hswebframework.web.system.authorization.defaults.service.reactive; +import com.google.common.cache.CacheBuilder; import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository; import org.hswebframework.web.authorization.Authentication; +import org.hswebframework.web.authorization.ReactiveAuthenticationInitializeService; import org.hswebframework.web.authorization.ReactiveAuthenticationManager; import org.hswebframework.web.authorization.User; +import org.hswebframework.web.authorization.events.AuthorizationInitializeEvent; import org.hswebframework.web.authorization.simple.PlainTextUsernamePasswordAuthenticationRequest; +import org.hswebframework.web.authorization.simple.SimpleAuthentication; +import org.hswebframework.web.cache.ReactiveCacheManager; +import org.hswebframework.web.cache.supports.GuavaReactiveCacheManager; import org.hswebframework.web.system.authorization.api.entity.ActionEntity; import org.hswebframework.web.system.authorization.api.entity.AuthorizationSettingEntity; import org.hswebframework.web.system.authorization.api.entity.PermissionEntity; import org.hswebframework.web.system.authorization.api.entity.UserEntity; +import org.hswebframework.web.system.authorization.api.event.ClearUserAuthorizationCacheEvent; import org.hswebframework.web.system.authorization.api.service.reactive.ReactiveUserService; -import org.hswebframework.web.system.authorization.defaults.service.DefaultReactiveUserService; +import org.hswebframework.web.system.authorization.defaults.service.AuthenticationInitializeProperties; +import org.hswebframework.web.system.authorization.defaults.service.DefaultReactiveAuthenticationInitializeService; +import org.hswebframework.web.system.authorization.defaults.service.DefaultReactiveAuthenticationManager; +import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.test.annotation.DirtiesContext; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.util.ReflectionTestUtils; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import java.util.Arrays; import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; @RunWith(SpringRunner.class) -@SpringBootTest(classes = {ReactiveTestApplication.class, DefaultReactiveUserService.class}) +@SpringBootTest(classes = ReactiveTestApplication.class) public class DefaultReactiveAuthenticationManagerTest { @Autowired @@ -39,6 +56,313 @@ public class DefaultReactiveAuthenticationManagerTest { @Autowired private ReactiveRepository settingRepository; + @Test + public void initializesPermissionsForNonDisabledUsersWithoutCache() { + String userId = "state-check"; + String permissionId = "non-disabled-state-permission"; + grantUserPermission(userId, permissionId); + UserEntity user = new UserEntity(); + user.setId(userId); + user.setStatus(UserEntity.STATUS_DISABLED); + AtomicInteger userReads = new AtomicInteger(); + AtomicInteger permissions = new AtomicInteger(); + AtomicInteger events = new AtomicInteger(); + ReactiveUserService users = mock(ReactiveUserService.class); + when(users.findById(userId)).thenReturn(Mono.defer(() -> { + userReads.incrementAndGet(); + return Mono.just(user); + })); + when(users.findById("missing")).thenReturn(Mono.defer(() -> { + userReads.incrementAndGet(); + return Mono.empty(); + })); + DefaultReactiveAuthenticationInitializeService initializer = createInitializer(users, permissions, events); + DefaultReactiveAuthenticationManager manager = createManager(users, initializer, null); + + manager.getByUserId("missing") + .as(StepVerifier::create) + .verifyComplete(); + manager.getByUserId(userId) + .as(StepVerifier::create) + .verifyComplete(); + Assert.assertEquals(2, userReads.get()); + Assert.assertEquals(0, permissions.get()); + Assert.assertEquals(0, events.get()); + + Byte[] readableStates = {UserEntity.STATUS_ENABLED, UserEntity.STATUS_LOCKED, null, (byte) 3}; + for (Byte status : readableStates) { + user.setStatus(status); + manager.getByUserId(userId) + .as(StepVerifier::create) + .assertNext(authentication -> { + assertUserStatus(authentication, userId, status); + Assert.assertTrue(authentication.hasPermission(permissionId, "query")); + }) + .verifyComplete(); + } + Assert.assertEquals(6, userReads.get()); + Assert.assertEquals(4, permissions.get()); + Assert.assertEquals(4, events.get()); + } + + @Test + public void reloadsEnabledUsersAfterCacheEvictionWithoutReadingWarmUsers() { + String userId = "cached-state-check"; + UserEntity user = new UserEntity(); + user.setId(userId); + user.setStatus(UserEntity.STATUS_ENABLED); + AtomicInteger userReads = new AtomicInteger(); + AtomicInteger permissions = new AtomicInteger(); + AtomicInteger events = new AtomicInteger(); + ReactiveUserService users = mock(ReactiveUserService.class); + when(users.findById(userId)).thenReturn(Mono.defer(() -> { + userReads.incrementAndGet(); + return Mono.just(user); + })); + when(users.findById("missing")).thenReturn(Mono.defer(() -> { + userReads.incrementAndGet(); + return Mono.empty(); + })); + DefaultReactiveAuthenticationInitializeService initializer = createInitializer(users, permissions, events); + DefaultReactiveAuthenticationManager manager = createManager( + users, initializer, new GuavaReactiveCacheManager(CacheBuilder.newBuilder())); + + for (int i = 0; i < 3; i++) { + manager.getByUserId(userId) + .as(StepVerifier::create) + .assertNext(authentication -> assertUserStatus(authentication, userId, UserEntity.STATUS_ENABLED)) + .verifyComplete(); + } + Assert.assertEquals(1, userReads.get()); + Assert.assertEquals(1, permissions.get()); + Assert.assertEquals(1, events.get()); + + user.setStatus(UserEntity.STATUS_DISABLED); + clearCache(manager, userId); + manager.getByUserId(userId) + .as(StepVerifier::create) + .verifyComplete(); + Assert.assertEquals(2, userReads.get()); + Assert.assertEquals(1, permissions.get()); + Assert.assertEquals(1, events.get()); + + user.setStatus(UserEntity.STATUS_LOCKED); + clearCache(manager, userId); + manager.getByUserId(userId) + .as(StepVerifier::create) + .assertNext(authentication -> assertUserStatus(authentication, userId, UserEntity.STATUS_LOCKED)) + .verifyComplete(); + manager.getByUserId(userId) + .as(StepVerifier::create) + .assertNext(authentication -> assertUserStatus(authentication, userId, UserEntity.STATUS_LOCKED)) + .verifyComplete(); + Assert.assertEquals(3, userReads.get()); + Assert.assertEquals(2, permissions.get()); + Assert.assertEquals(2, events.get()); + + user.setStatus(UserEntity.STATUS_ENABLED); + clearCache(manager, userId); + manager.getByUserId(userId) + .as(StepVerifier::create) + .assertNext(authentication -> assertUserStatus(authentication, userId, UserEntity.STATUS_ENABLED)) + .verifyComplete(); + manager.getByUserId(userId) + .as(StepVerifier::create) + .assertNext(authentication -> assertUserStatus(authentication, userId, UserEntity.STATUS_ENABLED)) + .verifyComplete(); + Assert.assertEquals(4, userReads.get()); + Assert.assertEquals(3, permissions.get()); + Assert.assertEquals(3, events.get()); + + user.setStatus(null); + clearCache(manager, userId); + manager.getByUserId(userId) + .as(StepVerifier::create) + .assertNext(authentication -> assertUserStatus(authentication, userId, null)) + .verifyComplete(); + Assert.assertEquals(5, userReads.get()); + Assert.assertEquals(4, permissions.get()); + Assert.assertEquals(4, events.get()); + + manager.getByUserId("missing") + .as(StepVerifier::create) + .verifyComplete(); + Assert.assertEquals(6, userReads.get()); + Assert.assertEquals(4, permissions.get()); + Assert.assertEquals(4, events.get()); + } + + @Test + public void passwordAuthenticationOnlyAcceptsEnabledUsers() { + String userId = "password-state-check"; + UserEntity user = new UserEntity(); + user.setId(userId); + user.setUsername(userId); + AtomicInteger userReads = new AtomicInteger(); + AtomicInteger permissions = new AtomicInteger(); + AtomicInteger events = new AtomicInteger(); + ReactiveUserService users = mock(ReactiveUserService.class); + when(users.findById(userId)).thenReturn(Mono.defer(() -> { + userReads.incrementAndGet(); + return Mono.just(user); + })); + when(users.findByUsernameAndPassword(userId, "password")) + .thenReturn(Mono.defer(() -> Mono.just(user))); + DefaultReactiveAuthenticationManager manager = createManager( + users, createInitializer(users, permissions, events), + new GuavaReactiveCacheManager(CacheBuilder.newBuilder())); + + Byte[] rejectedStates = {UserEntity.STATUS_DISABLED, UserEntity.STATUS_LOCKED, null, (byte) 3}; + for (Byte status : rejectedStates) { + user.setStatus(status); + manager.authenticate(Mono.just(new PlainTextUsernamePasswordAuthenticationRequest(userId, "password"))) + .as(StepVerifier::create) + .verifyComplete(); + } + Assert.assertEquals(0, userReads.get()); + Assert.assertEquals(0, permissions.get()); + Assert.assertEquals(0, events.get()); + + user.setStatus(UserEntity.STATUS_ENABLED); + manager.authenticate(Mono.just(new PlainTextUsernamePasswordAuthenticationRequest(userId, "password"))) + .as(StepVerifier::create) + .assertNext(authentication -> assertUserStatus(authentication, userId, UserEntity.STATUS_ENABLED)) + .verifyComplete(); + + user.setStatus(UserEntity.STATUS_LOCKED); + clearCache(manager, userId); + manager.getByUserId(userId) + .as(StepVerifier::create) + .assertNext(authentication -> assertUserStatus(authentication, userId, UserEntity.STATUS_LOCKED)) + .verifyComplete(); + manager.authenticate(Mono.just(new PlainTextUsernamePasswordAuthenticationRequest(userId, "password"))) + .as(StepVerifier::create) + .verifyComplete(); + + user.setStatus(UserEntity.STATUS_ENABLED); + clearCache(manager, userId); + for (int i = 0; i < 2; i++) { + manager.authenticate(Mono.just(new PlainTextUsernamePasswordAuthenticationRequest(userId, "password"))) + .as(StepVerifier::create) + .assertNext(authentication -> assertUserStatus(authentication, userId, UserEntity.STATUS_ENABLED)) + .verifyComplete(); + } + Assert.assertEquals(3, userReads.get()); + Assert.assertEquals(3, permissions.get()); + Assert.assertEquals(3, events.get()); + } + + @Test + public void propagatesUserLookupErrorsWithoutCache() { + String userId = "lookup-error"; + IllegalStateException failure = new IllegalStateException("database unavailable"); + ReactiveUserService users = mock(ReactiveUserService.class); + AtomicInteger permissions = new AtomicInteger(); + AtomicInteger events = new AtomicInteger(); + DefaultReactiveAuthenticationInitializeService initializer = createInitializer(users, permissions, events); + when(users.findById(userId)).thenReturn(Mono.error(failure)); + DefaultReactiveAuthenticationManager manager = createManager(users, initializer, null); + + manager.getByUserId(userId) + .as(StepVerifier::create) + .expectErrorSatisfies(error -> Assert.assertSame(failure, error)) + .verify(); + Assert.assertEquals(0, permissions.get()); + Assert.assertEquals(0, events.get()); + } + + @Test + public void delegatesToCustomInitializerWithoutPersistentUserLookup() { + String userId = "external-user"; + ReactiveUserService users = mock(ReactiveUserService.class); + ReactiveAuthenticationInitializeService initializer = mock(ReactiveAuthenticationInitializeService.class); + Authentication authentication = mock(Authentication.class); + when(initializer.initUserAuthorization(userId)).thenReturn(Mono.just(authentication)); + DefaultReactiveAuthenticationManager manager = createManager(users, initializer, null); + + manager.getByUserId(userId) + .as(StepVerifier::create) + .expectNext(authentication) + .verifyComplete(); + verify(initializer).initUserAuthorization(userId); + verifyNoInteractions(users); + } + + private static void assertUserStatus(Authentication authentication, String userId, Byte status) { + Assert.assertEquals(userId, authentication.getUser().getId()); + Assert.assertTrue(authentication.getUser().getOptions().containsKey(UserEntity.OPTION_STATUS)); + Assert.assertEquals(status, authentication.getUser().getOptions().get(UserEntity.OPTION_STATUS)); + } + + private void grantUserPermission(String userId, String permissionId) { + permissionRepository.newInstance() + .map(permission -> { + permission.setId(permissionId); + permission.setName(permissionId); + permission.setActions(Collections.singletonList( + ActionEntity.builder().action("query").describe("query").build())); + permission.setStatus((byte) 1); + return permission; + }) + .as(permissionRepository::insert) + .as(StepVerifier::create) + .expectNext(1) + .verifyComplete(); + settingRepository.newInstance() + .map(setting -> { + setting.setPermission(permissionId); + setting.setActions(Collections.singleton("query")); + setting.setDimensionType("user"); + setting.setDimensionTarget(userId); + setting.setState((byte) 1); + return setting; + }) + .as(settingRepository::insert) + .as(StepVerifier::create) + .expectNext(1) + .verifyComplete(); + } + + private DefaultReactiveAuthenticationInitializeService createInitializer( + ReactiveUserService users, AtomicInteger permissions, AtomicInteger events) { + DefaultReactiveAuthenticationInitializeService initializer = new DefaultReactiveAuthenticationInitializeService() { + @Override + protected Mono initPermission(SimpleAuthentication authentication) { + permissions.incrementAndGet(); + return super.initPermission(authentication); + } + }; + ApplicationEventPublisher publisher = event -> { + Assert.assertTrue(event instanceof AuthorizationInitializeEvent); + events.incrementAndGet(); + }; + ReflectionTestUtils.setField(initializer, "userService", users); + ReflectionTestUtils.setField(initializer, "settingRepository", settingRepository); + ReflectionTestUtils.setField(initializer, "permissionRepository", permissionRepository); + ReflectionTestUtils.setField(initializer, "properties", new AuthenticationInitializeProperties()); + ReflectionTestUtils.setField(initializer, "eventPublisher", publisher); + return initializer; + } + + private static DefaultReactiveAuthenticationManager createManager( + ReactiveUserService users, + ReactiveAuthenticationInitializeService initializer, + ReactiveCacheManager caches) { + DefaultReactiveAuthenticationManager manager = new DefaultReactiveAuthenticationManager(); + ReflectionTestUtils.setField(manager, "reactiveUserService", users); + ReflectionTestUtils.setField(manager, "initializeService", initializer); + ReflectionTestUtils.setField(manager, "cacheManager", caches); + return manager; + } + + private static void clearCache(DefaultReactiveAuthenticationManager manager, String userId) { + ClearUserAuthorizationCacheEvent event = ClearUserAuthorizationCacheEvent.of(userId).useAsync(); + manager.handleClearAuthCache(event); + event.getAsync() + .as(StepVerifier::create) + .verifyComplete(); + } + @Test public void test() { UserEntity entity = new UserEntity(); @@ -110,4 +434,4 @@ public void test() { } -} \ No newline at end of file +} diff --git a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/DefaultReactiveUserServiceTest.java b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/DefaultReactiveUserServiceTest.java index eeb66b885..4bd78be6f 100644 --- a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/DefaultReactiveUserServiceTest.java +++ b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/DefaultReactiveUserServiceTest.java @@ -3,13 +3,11 @@ import org.hswebframework.web.exception.ValidationException; import org.hswebframework.web.system.authorization.api.entity.UserEntity; import org.hswebframework.web.system.authorization.api.service.reactive.ReactiveUserService; -import org.hswebframework.web.system.authorization.defaults.service.DefaultReactiveUserService; import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; @@ -18,7 +16,7 @@ import java.util.function.Supplier; @RunWith(SpringRunner.class) -@SpringBootTest(classes = {ReactiveTestApplication.class, DefaultReactiveUserService.class}) +@SpringBootTest(classes = ReactiveTestApplication.class) public class DefaultReactiveUserServiceTest { @Autowired @@ -95,4 +93,4 @@ public void testCrud() { } -} \ No newline at end of file +} diff --git a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/WebFluxPermissionControllerTest.java b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/WebFluxPermissionControllerTest.java index 45c091ae9..c0f6f25d2 100644 --- a/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/WebFluxPermissionControllerTest.java +++ b/hsweb-system/hsweb-system-authorization/hsweb-system-authorization-default/src/test/java/org/hswebframework/web/system/authorization/defaults/service/reactive/WebFluxPermissionControllerTest.java @@ -1,6 +1,7 @@ package org.hswebframework.web.system.authorization.defaults.service.reactive; import org.hswebframework.web.crud.configuration.EasyormConfiguration; +import org.hswebframework.web.crud.configuration.EntityFactoryConfiguration; import org.hswebframework.web.crud.configuration.JdbcSqlExecutorConfiguration; import org.hswebframework.web.crud.configuration.R2dbcSqlExecutorConfiguration; import org.hswebframework.web.system.authorization.defaults.configuration.AuthorizationServiceAutoConfiguration; @@ -20,11 +21,13 @@ import org.springframework.transaction.annotation.EnableTransactionManagement; @RunWith(SpringRunner.class) -@WebFluxTest(WebFluxPermissionController.class) +@WebFluxTest(value = WebFluxPermissionController.class, + properties = "spring.r2dbc.generate-unique-name=true") @ImportAutoConfiguration(value = { AuthorizationWebAutoConfiguration.class, AuthorizationServiceAutoConfiguration.class, EasyormConfiguration.class, + EntityFactoryConfiguration.class, R2dbcSqlExecutorConfiguration.class, R2dbcAutoConfiguration.class, R2dbcTransactionManagerAutoConfiguration.class },exclude = {