This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 9696a3abf3 feat(cache): coalesce concurrent cache misses into one
upstream fill (#7969)
9696a3abf3 is described below
commit 9696a3abf3ac92a3025aae2093e94674969e8946
Author: hanzhe-one <[email protected]>
AuthorDate: Fri Oct 23 10:05:03 2026 +0800
feat(cache): coalesce concurrent cache misses into one upstream fill (#7969)
* feat(cache): coalesce concurrent cache misses into one upstream fill
Concurrent requests that miss the same cache key each forward to the
upstream, so a burst on one cold key produces one upstream call per
request. Add a per-gateway coordinator that lets the first request fill
the key and has later requests on the same key replay the captured
response instead.
A follower never subscribes to the leader's response publisher and never
shares a DataBuffer: it decodes its own copy of the response packet into
its own response. When the leader cannot produce a reusable response, or
a follower waits too long, runs out of waiter slots, or would receive a
response above the configured size, the follower fills the key itself,
which is the behaviour without coalescing.
Keys are exactly the existing response cache keys, so coalescing never
widens cache eligibility or merges responses the cache keeps apart. The
feature is off by default and bounded by filling keys, waiters per key,
wait time, fill time and shared response size. Coordination is per
gateway instance; replicas may still fill the same key independently.
Stale-while-revalidate and configuration generations are follow-ups.
* test(cache): cover the remaining single-flight branches
Add the three branches the review called out as uncovered:
- `maxKeys` capacity: with one key slot taken, a second key forwards
instead of being folded into another fill.
- A leader whose chain fails, where the follower is resolved by the
release path rather than by a published packet.
- Follower status replay: a 201 and a 404 leader both reach the follower
with their status and headers, not just the body.
Also sweep expired fills from `follow()`, so a fill whose leader vanished
on an idle key is released by time instead of waiting for the next
request to reach the key limit, and note the soft nature of the `maxKeys`
check and the coordinator swap on config reload.
Add the operator-facing RELEASE-NOTES entry under Unreleased.
---
RELEASE-NOTES.md | 18 +
.../apache/shenyu/plugin/cache/CachePlugin.java | 92 ++++-
.../plugin/cache/cache/ResponseSingleFlight.java | 218 ++++++++++
.../shenyu/plugin/cache/config/CacheConfig.java | 138 +++++++
.../cache/handler/CachePluginDataHandler.java | 22 +
.../shenyu/plugin/cache/utils/CacheUtils.java | 12 +
.../shenyu/plugin/cache/CacheSingleFlightTest.java | 450 +++++++++++++++++++++
7 files changed, 943 insertions(+), 7 deletions(-)
diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md
index 443a806f51..99f25dc346 100644
--- a/RELEASE-NOTES.md
+++ b/RELEASE-NOTES.md
@@ -28,6 +28,24 @@
### Behavior Changes
+- The cache plugin accepts opt-in `singleFlightEnabled` coalescing. When
enabled,
+ concurrent requests that miss the same response cache key share one upstream
+ fill on that gateway instance instead of one per request, and the followers
+ replay the captured bytes as a cache hit would. A response that the cache
would
+ not have stored, a fill that errors, a follower that waits longer than
+ `singleFlightWaitTimeoutMillis`, a full waiter set and a response above
+ `singleFlightMaxResponseBytes` all fall back to forwarding that request, so
+ coalescing never widens cache eligibility or merges responses kept apart by
+ cookie, `Vary` or cache-control rules. Bounds are set by
+ `singleFlightMaxKeys`, `singleFlightMaxWaiters`,
`singleFlightWaitTimeoutMillis`,
+ `singleFlightFillTimeoutMillis` and `singleFlightMaxResponseBytes`. The
feature
+ is disabled by default and absent configuration preserves the previous
+ one-forward-per-request behavior. Coordination is per gateway instance:
replicas
+ still fill the same key independently and no distributed lock is taken. Set
+ `singleFlightWaitTimeoutMillis` to at least the upstream p99, otherwise slow
+ upstreams finish after followers have already given up and the burst is not
+ collapsed. (#7932)
+
- Agent Gateway accepts opt-in `mcp.httpTools` plans for fixed existing gateway
HTTP routes. GET/query and POST/query or flat scalar JSON-body mappings use
request-local plugin cursors, attributes and captured responses; upstream
diff --git
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/CachePlugin.java
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/CachePlugin.java
index ace097d458..6387289bb7 100644
---
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/CachePlugin.java
+++
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/CachePlugin.java
@@ -28,6 +28,7 @@ import org.apache.shenyu.plugin.api.ShenyuPluginChain;
import org.apache.shenyu.plugin.base.AbstractShenyuPlugin;
import org.apache.shenyu.plugin.base.utils.CacheKeyUtils;
import org.apache.shenyu.plugin.cache.cache.ApplicationConfigCache;
+import org.apache.shenyu.plugin.cache.cache.ResponseSingleFlight;
import org.apache.shenyu.plugin.cache.handler.CachePluginDataHandler;
import org.apache.shenyu.plugin.cache.utils.CacheUtils;
import org.reactivestreams.Publisher;
@@ -106,10 +107,55 @@ public class CachePlugin extends AbstractShenyuPlugin {
if (dataFlg.isPresent()) {
return
dataFlg.get().writeTo(exchange.getResponse());
}
- return chain.execute(exchange.mutate().response(new
CacheHttpResponse(exchange, cacheRuleHandle, selector.getId(),
responseKey)).build());
+ return fillCache(exchange, chain, cacheRuleHandle,
selector.getId(), responseKey);
});
}
- return chain.execute(exchange.mutate().response(new
CacheHttpResponse(exchange, cacheRuleHandle, selector.getId(),
responseKey)).build());
+ return fillCache(exchange, chain, cacheRuleHandle, selector.getId(),
responseKey);
+ }
+
+ /**
+ * Fill a missing key, sharing one upstream call between the requests that
miss on it at the same time.
+ *
+ * @param exchange the exchange
+ * @param chain the plugin chain
+ * @param cacheRuleHandle the matched rule handle
+ * @param selectorId the matched selector
+ * @param responseKey the response cache key
+ * @return the response completion
+ */
+ private Mono<Void> fillCache(final ServerWebExchange exchange, final
ShenyuPluginChain chain,
+ final CacheRuleHandle cacheRuleHandle, final
String selectorId, final String responseKey) {
+ final ResponseSingleFlight singleFlight = CacheUtils.getSingleFlight();
+ final Optional<ResponseSingleFlight.Fill> owned =
singleFlight.lead(responseKey);
+ if (owned.isEmpty()) {
+ final Optional<Mono<Optional<byte[]>>> followed =
singleFlight.follow(responseKey);
+ if (followed.isPresent()) {
+ return followed.get().flatMap(packet -> packet
+ .flatMap(CachedResponse::decode)
+ .map(response ->
response.writeTo(exchange.getResponse()))
+ .orElseGet(() -> forward(exchange, chain,
cacheRuleHandle, selectorId, responseKey)));
+ }
+ }
+ final ResponseSingleFlight.Fill fill = owned.orElse(null);
+ final Mono<Void> forward = forward(exchange, chain, cacheRuleHandle,
selectorId, responseKey, fill);
+ if (Objects.isNull(fill)) {
+ return forward;
+ }
+ // Release the key on error and cancellation. A successful write has
already published or released
+ // the fill, and releasing twice is a no-op.
+ return forward.doFinally(ignored -> fill.release());
+ }
+
+ private Mono<Void> forward(final ServerWebExchange exchange, final
ShenyuPluginChain chain,
+ final CacheRuleHandle cacheRuleHandle, final
String selectorId, final String responseKey) {
+ return forward(exchange, chain, cacheRuleHandle, selectorId,
responseKey, null);
+ }
+
+ private Mono<Void> forward(final ServerWebExchange exchange, final
ShenyuPluginChain chain,
+ final CacheRuleHandle cacheRuleHandle, final
String selectorId,
+ final String responseKey, final
ResponseSingleFlight.Fill fill) {
+ return chain.execute(exchange.mutate().response(
+ new CacheHttpResponse(exchange, cacheRuleHandle, selectorId,
responseKey, fill)).build());
}
/**
@@ -152,13 +198,22 @@ public class CachePlugin extends AbstractShenyuPlugin {
private final CacheRuleHandle cacheRuleHandle;
+ private final ResponseSingleFlight.Fill fill;
+
CacheHttpResponse(final ServerWebExchange exchange,
final CacheRuleHandle cacheRuleHandle,
final String selectorId, final String responseKey) {
+ this(exchange, cacheRuleHandle, selectorId, responseKey, null);
+ }
+
+ CacheHttpResponse(final ServerWebExchange exchange,
+ final CacheRuleHandle cacheRuleHandle,
+ final String selectorId, final String responseKey,
final ResponseSingleFlight.Fill fill) {
super(exchange.getResponse());
this.responseKey = responseKey;
this.cacheRuleHandle = cacheRuleHandle;
this.selectorId = selectorId;
+ this.fill = fill;
}
@Override
@@ -194,26 +249,49 @@ public class CachePlugin extends AbstractShenyuPlugin {
} else {
getHeaders().setContentLength(body.length);
}
- AtomicReference<CachedResponse> captured = new
AtomicReference<>();
+ AtomicReference<Optional<CachedResponse>> captured = new
AtomicReference<>(Optional.empty());
getDelegate().beforeCommit(() -> {
try {
- captured.set(CachedResponse.capture(getDelegate(),
body).orElse(null));
+ captured.set(CachedResponse.capture(getDelegate(),
body));
} catch (RuntimeException ex) {
LOG.warn("Failed to capture response cache metadata",
ex);
}
return Mono.empty();
});
Mono<Void> write = body.length == 0 ? super.setComplete() :
super.writeWith(Mono.just(bufferFactory().wrap(body)));
- return write.doOnSuccess(ignored ->
Optional.ofNullable(captured.get()).ifPresent(this::cacheResponse));
+ return write.doOnSuccess(ignored -> publish(captured.get()));
});
}
- private void cacheResponse(final CachedResponse response) {
+ /**
+ * Store the captured response and hand the same bytes to the requests
waiting on this fill.
+ *
+ * @param response the captured response, empty when this response
cannot be reused
+ */
+ private void publish(final Optional<CachedResponse> response) {
+ if (response.isEmpty()) {
+ releaseFill();
+ return;
+ }
+ final byte[] packet = response.get().encode();
+ storeCache(packet);
+ if (Objects.nonNull(fill)) {
+ fill.complete(Optional.of(packet));
+ }
+ }
+
+ private void releaseFill() {
+ if (Objects.nonNull(fill)) {
+ fill.release();
+ }
+ }
+
+ private void storeCache(final byte[] packet) {
try {
ICache selectorCache =
ApplicationConfigCache.getInstance().get(selectorId);
ICache cache = Objects.isNull(selectorCache) ?
CacheUtils.getCache() : selectorCache;
if (Objects.nonNull(cache)) {
- cache.cacheData(responseKey, response.encode(),
cacheRuleHandle.getTimeoutSeconds())
+ cache.cacheData(responseKey, packet,
cacheRuleHandle.getTimeoutSeconds())
.subscribeOn(Schedulers.boundedElastic())
.subscribe(ignored -> { }, error ->
LOG.warn("Failed to store response cache entry", error));
}
diff --git
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/cache/ResponseSingleFlight.java
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/cache/ResponseSingleFlight.java
new file mode 100644
index 0000000000..6a6a8e09cb
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/cache/ResponseSingleFlight.java
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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
+ *
+ * http://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 org.apache.shenyu.plugin.cache.cache;
+
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
+import java.time.Duration;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Coalesces concurrent cache misses for one key into a single upstream fill
on this gateway instance.
+ *
+ * <p>The first request for a key leads the fill and executes the plugin
chain; later requests for the
+ * same key follow it and replay the captured response bytes once the leader
publishes them. A follower
+ * never subscribes to the leader's response publisher and never shares a
{@code DataBuffer}: it decodes
+ * its own copy of the bytes into its own response. When the leader cannot
produce a reusable response,
+ * or the follower waits too long, the follower fills the key itself, which is
the behaviour without
+ * coalescing.</p>
+ *
+ * <p>Coordination is per gateway instance. Replicas may still fill the same
key independently, and no
+ * distributed lock is taken. Keys are exactly the response cache keys, so
coalescing never widens cache
+ * eligibility or merges responses that the cache would have kept apart.</p>
+ */
+public final class ResponseSingleFlight {
+
+ private static final ResponseSingleFlight DISABLED = new
ResponseSingleFlight(0, 0, 0L, 0L, 0L);
+
+ private final int maxKeys;
+
+ private final int maxWaiters;
+
+ private final long maxResponseBytes;
+
+ private final Duration waitTimeout;
+
+ private final long fillTimeoutMillis;
+
+ private final ConcurrentMap<String, Fill> fills = new
ConcurrentHashMap<>();
+
+ /**
+ * Instantiates a coordinator.
+ *
+ * @param maxKeys the maximum number of keys filling at once, zero or less
disables coalescing
+ * @param maxWaiters the maximum number of requests waiting on one key
+ * @param waitTimeoutMillis how long a request waits for another request's
fill
+ * @param fillTimeoutMillis how long a key stays owned when its leader
never completes
+ * @param maxResponseBytes the largest response shared with waiting
requests, zero or less shares none
+ */
+ public ResponseSingleFlight(final int maxKeys, final int maxWaiters, final
long waitTimeoutMillis,
+ final long fillTimeoutMillis, final long
maxResponseBytes) {
+ this.maxKeys = maxKeys;
+ this.maxWaiters = maxWaiters;
+ this.waitTimeout = Duration.ofMillis(waitTimeoutMillis);
+ this.fillTimeoutMillis = fillTimeoutMillis;
+ this.maxResponseBytes = maxResponseBytes;
+ }
+
+ /**
+ * A coordinator that never coalesces, used when the feature is switched
off.
+ *
+ * @return the disabled coordinator
+ */
+ public static ResponseSingleFlight disabled() {
+ return DISABLED;
+ }
+
+ /**
+ * Try to lead the fill of a key.
+ *
+ * @param key the response cache key
+ * @return the owned fill, or empty when the key is already filling or
this instance is at capacity
+ */
+ public Optional<Fill> lead(final String key) {
+ if (maxKeys <= 0) {
+ return Optional.empty();
+ }
+ // The size check and putIfAbsent are not atomic, so concurrent
leaders can exceed maxKeys by the
+ // number of requests racing here. This bounds capacity; it is not an
exact count.
+ if (fills.size() >= maxKeys) {
+ expireStale();
+ if (fills.size() >= maxKeys) {
+ return Optional.empty();
+ }
+ }
+ final Fill candidate = new Fill(this, key, System.nanoTime());
+ final Fill existing = fills.putIfAbsent(key, candidate);
+ return Objects.isNull(existing) ? Optional.of(candidate) :
Optional.empty();
+ }
+
+ /**
+ * Follow the fill another request leads for a key.
+ *
+ * @param key the response cache key
+ * @return a signal resolving to the reusable response packet, or empty
when there is nothing to follow
+ */
+ public Optional<Mono<Optional<byte[]>>> follow(final String key) {
+ // Sweep here as well as in lead(), so a fill whose leader vanished on
an otherwise idle key is
+ // released by time rather than waiting for the next request to reach
the key limit.
+ expireStale();
+ // A fill that has already published stays visible until it is
removed, so a request arriving in
+ // that window still replays the packet instead of filling the key
again.
+ final Fill fill = fills.get(key);
+ if (Objects.isNull(fill)) {
+ return Optional.empty();
+ }
+ if (fill.waiters.incrementAndGet() > maxWaiters) {
+ fill.waiters.decrementAndGet();
+ return Optional.empty();
+ }
+ return Optional.of(fill.signal.asMono()
+ .defaultIfEmpty(Optional.empty())
+ .timeout(waitTimeout)
+ .onErrorReturn(Optional.empty())
+ .doFinally(ignored -> fill.waiters.decrementAndGet()));
+ }
+
+ /**
+ * Publish a reusable response packet and release the key.
+ *
+ * <p>Only the first call for a fill takes effect, so a leader that
already published a packet is not
+ * affected by a later cancellation.</p>
+ *
+ * @param fill the owned fill
+ * @param packet the encoded response, empty when the response cannot be
reused
+ */
+ public void complete(final Fill fill, final Optional<byte[]> packet) {
+ if (!fill.finished.compareAndSet(false, true)) {
+ return;
+ }
+ final Optional<byte[]> shared = maxResponseBytes <= 0
+ ? Optional.empty()
+ : packet.filter(bytes -> bytes.length <= maxResponseBytes);
+ // Publish before releasing the key so a request arriving in between
still replays this fill.
+ fill.signal.tryEmitValue(shared);
+ fills.remove(fill.key, fill);
+ }
+
+ /**
+ * Release the key without a reusable response, for example when the fill
failed or was cancelled.
+ *
+ * @param fill the owned fill
+ */
+ public void release(final Fill fill) {
+ complete(fill, Optional.empty());
+ }
+
+ /**
+ * Release keys whose leader never completed, so one stuck request cannot
hold a key forever.
+ */
+ private void expireStale() {
+ final long now = System.nanoTime();
+ final long timeoutNanos =
Duration.ofMillis(fillTimeoutMillis).toNanos();
+ fills.values().stream()
+ .filter(fill -> now - fill.createdAtNanos > timeoutNanos)
+ .forEach(this::release);
+ }
+
+ /**
+ * One upstream fill owned by a single request.
+ */
+ public static final class Fill {
+
+ private final ResponseSingleFlight owner;
+
+ private final String key;
+
+ private final long createdAtNanos;
+
+ private final Sinks.One<Optional<byte[]>> signal = Sinks.one();
+
+ private final AtomicInteger waiters = new AtomicInteger();
+
+ private final AtomicBoolean finished = new AtomicBoolean();
+
+ private Fill(final ResponseSingleFlight owner, final String key, final
long createdAtNanos) {
+ this.owner = owner;
+ this.key = key;
+ this.createdAtNanos = createdAtNanos;
+ }
+
+ /**
+ * Publish a reusable response packet and release the key.
+ *
+ * @param packet the encoded response, empty when the response cannot
be reused
+ */
+ public void complete(final Optional<byte[]> packet) {
+ owner.complete(this, packet);
+ }
+
+ /**
+ * Release the key without a reusable response.
+ */
+ public void release() {
+ owner.release(this);
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/config/CacheConfig.java
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/config/CacheConfig.java
index bf74315114..b24b98a5cd 100644
---
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/config/CacheConfig.java
+++
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/config/CacheConfig.java
@@ -34,6 +34,36 @@ public class CacheConfig {
*/
private String config;
+ /**
+ * whether concurrent misses on one key share a single upstream fill, off
by default.
+ */
+ private boolean singleFlightEnabled;
+
+ /**
+ * the maximum number of keys filling at once on this gateway.
+ */
+ private int singleFlightMaxKeys = 1024;
+
+ /**
+ * the maximum number of requests waiting on one filling key.
+ */
+ private int singleFlightMaxWaiters = 64;
+
+ /**
+ * how long a request waits for another request's fill, in milliseconds.
+ */
+ private long singleFlightWaitTimeoutMillis = 1000L;
+
+ /**
+ * how long a key stays owned when its leader never completes, in
milliseconds.
+ */
+ private long singleFlightFillTimeoutMillis = 30000L;
+
+ /**
+ * the largest response shared with waiting requests, in bytes.
+ */
+ private long singleFlightMaxResponseBytes = 1048576L;
+
/**
* Get cache type.
* @return the cache type
@@ -66,6 +96,114 @@ public class CacheConfig {
this.config = config;
}
+ /**
+ * Whether concurrent misses on one key share a single upstream fill.
+ *
+ * @return true when coalescing is enabled
+ */
+ public boolean isSingleFlightEnabled() {
+ return singleFlightEnabled;
+ }
+
+ /**
+ * Set whether concurrent misses on one key share a single upstream fill.
+ *
+ * @param singleFlightEnabled true to enable coalescing
+ */
+ public void setSingleFlightEnabled(final boolean singleFlightEnabled) {
+ this.singleFlightEnabled = singleFlightEnabled;
+ }
+
+ /**
+ * Get the maximum number of keys filling at once.
+ *
+ * @return the maximum number of filling keys
+ */
+ public int getSingleFlightMaxKeys() {
+ return singleFlightMaxKeys;
+ }
+
+ /**
+ * Set the maximum number of keys filling at once.
+ *
+ * @param singleFlightMaxKeys the maximum number of filling keys
+ */
+ public void setSingleFlightMaxKeys(final int singleFlightMaxKeys) {
+ this.singleFlightMaxKeys = singleFlightMaxKeys;
+ }
+
+ /**
+ * Get the maximum number of requests waiting on one filling key.
+ *
+ * @return the maximum number of waiters
+ */
+ public int getSingleFlightMaxWaiters() {
+ return singleFlightMaxWaiters;
+ }
+
+ /**
+ * Set the maximum number of requests waiting on one filling key.
+ *
+ * @param singleFlightMaxWaiters the maximum number of waiters
+ */
+ public void setSingleFlightMaxWaiters(final int singleFlightMaxWaiters) {
+ this.singleFlightMaxWaiters = singleFlightMaxWaiters;
+ }
+
+ /**
+ * Get how long a request waits for another request's fill.
+ *
+ * @return the wait timeout in milliseconds
+ */
+ public long getSingleFlightWaitTimeoutMillis() {
+ return singleFlightWaitTimeoutMillis;
+ }
+
+ /**
+ * Set how long a request waits for another request's fill.
+ *
+ * @param singleFlightWaitTimeoutMillis the wait timeout in milliseconds
+ */
+ public void setSingleFlightWaitTimeoutMillis(final long
singleFlightWaitTimeoutMillis) {
+ this.singleFlightWaitTimeoutMillis = singleFlightWaitTimeoutMillis;
+ }
+
+ /**
+ * Get how long a key stays owned when its leader never completes.
+ *
+ * @return the fill timeout in milliseconds
+ */
+ public long getSingleFlightFillTimeoutMillis() {
+ return singleFlightFillTimeoutMillis;
+ }
+
+ /**
+ * Set how long a key stays owned when its leader never completes.
+ *
+ * @param singleFlightFillTimeoutMillis the fill timeout in milliseconds
+ */
+ public void setSingleFlightFillTimeoutMillis(final long
singleFlightFillTimeoutMillis) {
+ this.singleFlightFillTimeoutMillis = singleFlightFillTimeoutMillis;
+ }
+
+ /**
+ * Get the largest response shared with waiting requests.
+ *
+ * @return the maximum shared response size in bytes
+ */
+ public long getSingleFlightMaxResponseBytes() {
+ return singleFlightMaxResponseBytes;
+ }
+
+ /**
+ * Set the largest response shared with waiting requests.
+ *
+ * @param singleFlightMaxResponseBytes the maximum shared response size in
bytes
+ */
+ public void setSingleFlightMaxResponseBytes(final long
singleFlightMaxResponseBytes) {
+ this.singleFlightMaxResponseBytes = singleFlightMaxResponseBytes;
+ }
+
@Override
public boolean equals(final Object o) {
if (this == o) {
diff --git
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/handler/CachePluginDataHandler.java
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/handler/CachePluginDataHandler.java
index 346c81021b..93850a833f 100644
---
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/handler/CachePluginDataHandler.java
+++
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/handler/CachePluginDataHandler.java
@@ -33,6 +33,7 @@ import org.apache.shenyu.plugin.base.utils.CacheKeyUtils;
import org.apache.shenyu.plugin.cache.ICache;
import org.apache.shenyu.plugin.cache.ICacheBuilder;
import org.apache.shenyu.plugin.cache.cache.ApplicationConfigCache;
+import org.apache.shenyu.plugin.cache.cache.ResponseSingleFlight;
import org.apache.shenyu.plugin.cache.config.CacheConfig;
import org.apache.shenyu.plugin.cache.utils.CacheUtils;
import org.apache.shenyu.spi.ExtensionLoader;
@@ -77,6 +78,7 @@ public class CachePluginDataHandler implements
PluginDataHandler {
return;
}
Singleton.INST.single(CacheConfig.class, cacheConfig);
+ Singleton.INST.single(ResponseSingleFlight.class,
buildSingleFlight(cacheConfig));
final ICacheBuilder cacheBuilder =
ExtensionLoader.getExtensionLoader(ICacheBuilder.class).getJoin(cacheConfig.getCacheType());
final ICache lastCache = CacheUtils.getCache();
ApplicationConfigCache.getInstance().invalidateAll();
@@ -127,12 +129,32 @@ public class CachePluginDataHandler implements
PluginDataHandler {
return PluginEnum.CACHE.getName();
}
+ /**
+ * Build the miss coordinator from the plugin configuration.
+ *
+ * <p>A configuration reload installs a new coordinator. Fills already
running keep their own
+ * coordinator and still signal their waiters, so nothing hangs; a request
that follows just after
+ * the swap starts a new fill on the new instance.</p>
+ *
+ * @param cacheConfig the plugin configuration
+ * @return the coordinator, disabled unless configured otherwise
+ */
+ private ResponseSingleFlight buildSingleFlight(final CacheConfig
cacheConfig) {
+ if (!cacheConfig.isSingleFlightEnabled()) {
+ return ResponseSingleFlight.disabled();
+ }
+ return new ResponseSingleFlight(cacheConfig.getSingleFlightMaxKeys(),
cacheConfig.getSingleFlightMaxWaiters(),
+ cacheConfig.getSingleFlightWaitTimeoutMillis(),
cacheConfig.getSingleFlightFillTimeoutMillis(),
+ cacheConfig.getSingleFlightMaxResponseBytes());
+ }
+
/**
* close the cache if you need.
*/
private void closeCacheIfNeed() {
ICache lastCache = CacheUtils.getCache();
ApplicationConfigCache.getInstance().invalidateAll();
+ Singleton.INST.single(ResponseSingleFlight.class,
ResponseSingleFlight.disabled());
this.closeCache(lastCache);
}
diff --git
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/utils/CacheUtils.java
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/utils/CacheUtils.java
index 65dc18f3f9..79226c94fc 100644
---
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/utils/CacheUtils.java
+++
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/utils/CacheUtils.java
@@ -21,10 +21,12 @@ import org.apache.commons.codec.digest.DigestUtils;
import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.plugin.cache.ICache;
+import org.apache.shenyu.plugin.cache.cache.ResponseSingleFlight;
import org.springframework.web.server.ServerWebExchange;
import java.net.URI;
import java.util.Arrays;
+import java.util.Objects;
/**
* CacheUtils.
@@ -82,4 +84,14 @@ public final class CacheUtils {
public static ICache getCache() {
return Singleton.INST.get(ICache.class);
}
+
+ /**
+ * get the miss coordinator of this gateway.
+ *
+ * @return the coordinator, never null
+ */
+ public static ResponseSingleFlight getSingleFlight() {
+ ResponseSingleFlight singleFlight =
Singleton.INST.get(ResponseSingleFlight.class);
+ return Objects.isNull(singleFlight) ? ResponseSingleFlight.disabled()
: singleFlight;
+ }
}
diff --git
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/test/java/org/apache/shenyu/plugin/cache/CacheSingleFlightTest.java
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/test/java/org/apache/shenyu/plugin/cache/CacheSingleFlightTest.java
new file mode 100644
index 0000000000..f621fab7f0
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/test/java/org/apache/shenyu/plugin/cache/CacheSingleFlightTest.java
@@ -0,0 +1,450 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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
+ *
+ * http://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 org.apache.shenyu.plugin.cache;
+
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.dto.convert.rule.impl.CacheRuleHandle;
+import org.apache.shenyu.common.dto.convert.selector.CacheUpstream;
+import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.common.utils.Singleton;
+import org.apache.shenyu.plugin.api.ShenyuPluginChain;
+import org.apache.shenyu.plugin.api.result.DefaultShenyuResult;
+import org.apache.shenyu.plugin.api.result.ShenyuResult;
+import org.apache.shenyu.plugin.api.utils.SpringBeanUtils;
+import org.apache.shenyu.plugin.base.utils.CacheKeyUtils;
+import org.apache.shenyu.plugin.cache.cache.ApplicationConfigCache;
+import org.apache.shenyu.plugin.cache.cache.ResponseSingleFlight;
+import org.apache.shenyu.plugin.cache.handler.CachePluginDataHandler;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.context.ApplicationContext;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatusCode;
+import org.springframework.http.MediaType;
+import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
+import org.springframework.mock.web.server.MockServerWebExchange;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.Disposable;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+import reactor.test.StepVerifier;
+
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Verifies that concurrent misses on one key share a single upstream fill.
+ */
+class CacheSingleFlightTest {
+
+ private static final String SELECTOR = "cache-single-flight-selector";
+
+ private static final String URI =
"http://gateway.example/cache-single-flight";;
+
+ private final CachePlugin plugin = new CachePlugin();
+
+ private final SelectorData selector = new SelectorData();
+
+ private final RuleData rule = new RuleData();
+
+ private ICache cache;
+
+ @BeforeEach
+ void setUp() {
+ selector.setId(SELECTOR);
+
selector.setHandle("{\"cacheType\":\"memory\",\"url\":\"memory://single-flight\"}");
+ rule.setSelectorId(SELECTOR);
+ rule.setId("cache-single-flight-rule");
+
CachePluginDataHandler.CACHED_HANDLE.get().cachedHandle(CacheKeyUtils.INST.getKey(rule),
new CacheRuleHandle());
+ cache = ApplicationConfigCache.getInstance().init(SELECTOR,
GsonUtils.getInstance().fromJson(selector.getHandle(), CacheUpstream.class));
+ ApplicationContext context = mock(ApplicationContext.class);
+ when(context.getBean(ShenyuResult.class)).thenReturn(new
DefaultShenyuResult());
+ SpringBeanUtils.getInstance().setApplicationContext(context);
+ }
+
+ @AfterEach
+ void tearDown() {
+ ApplicationConfigCache.getInstance().invalidateCache(SELECTOR);
+
CachePluginDataHandler.CACHED_HANDLE.get().removeHandle(CacheKeyUtils.INST.getKey(rule));
+ Singleton.INST.single(ResponseSingleFlight.class,
ResponseSingleFlight.disabled());
+ SpringBeanUtils.getInstance().setApplicationContext(null);
+ }
+
+ @Test
+ void testConcurrentMissesShareOneFill() throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class, new
ResponseSingleFlight(64, 64, 5000L, 30000L, 1048576L));
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ ShenyuPluginChain chain = exchange -> {
+ calls.incrementAndGet();
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ leaderStarted.countDown();
+ return gate.asMono().then(write(exchange, "shared"));
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange follower = request();
+
+ StepVerifier.create(execute(follower, chain))
+ .then(() -> gate.tryEmitEmpty())
+ .verifyComplete();
+
+ assertEquals(1, calls.get());
+ assertEquals("shared",
follower.getResponse().getBodyAsString().block());
+ assertEquals("shared",
leader.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ @Test
+ void testDisabledCoalescingForwardsEveryRequest() throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class,
ResponseSingleFlight.disabled());
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ ShenyuPluginChain chain = exchange -> {
+ calls.incrementAndGet();
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ leaderStarted.countDown();
+ return gate.asMono().then(write(exchange, "body-" + calls.get()));
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange follower = request();
+
+ StepVerifier.create(execute(follower, chain))
+ .then(() -> gate.tryEmitEmpty())
+ .verifyComplete();
+
+ assertEquals(2, calls.get());
+ assertEquals("body-2",
follower.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ @Test
+ void testFollowerFillsWhenLeaderResponseCannotBeShared() throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class, new
ResponseSingleFlight(64, 64, 5000L, 30000L, 1048576L));
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ ShenyuPluginChain chain = exchange -> {
+ int call = calls.incrementAndGet();
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ if (call == 1) {
+
exchange.getResponse().getHeaders().add(HttpHeaders.SET_COOKIE,
"session=private");
+ leaderStarted.countDown();
+ return gate.asMono().then(write(exchange, "private"));
+ }
+ return write(exchange, "shared");
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange follower = request();
+
+ StepVerifier.create(execute(follower, chain))
+ .then(() -> gate.tryEmitEmpty())
+ .verifyComplete();
+
+ assertEquals(2, calls.get());
+ assertEquals("shared",
follower.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ @Test
+ void testFollowerFillsAfterWaitingTooLong() throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class, new
ResponseSingleFlight(64, 64, 50L, 30000L, 1048576L));
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ ShenyuPluginChain chain = exchange -> {
+ int call = calls.incrementAndGet();
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ if (call == 1) {
+ leaderStarted.countDown();
+ return gate.asMono().then(write(exchange, "slow"));
+ }
+ return write(exchange, "fast");
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange follower = request();
+
+ StepVerifier.create(execute(follower, chain)).verifyComplete();
+
+ assertEquals(2, calls.get());
+ assertEquals("fast",
follower.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ @Test
+ void testFollowerFillsWhenResponseExceedsShareLimit() throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class, new
ResponseSingleFlight(64, 64, 5000L, 30000L, 4L));
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ ShenyuPluginChain chain = exchange -> {
+ int call = calls.incrementAndGet();
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ if (call == 1) {
+ leaderStarted.countDown();
+ return gate.asMono().then(write(exchange,
"larger-than-limit"));
+ }
+ return write(exchange, "fast");
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange follower = request();
+
+ StepVerifier.create(execute(follower, chain))
+ .then(() -> gate.tryEmitEmpty())
+ .verifyComplete();
+
+ assertEquals(2, calls.get());
+ assertEquals("fast",
follower.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ @Test
+ void testFollowerFillsWhenNoWaiterSlotIsAvailable() throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class, new
ResponseSingleFlight(64, 0, 5000L, 30000L, 1048576L));
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ ShenyuPluginChain chain = exchange -> {
+ int call = calls.incrementAndGet();
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ if (call == 1) {
+ leaderStarted.countDown();
+ return gate.asMono().then(write(exchange, "slow"));
+ }
+ return write(exchange, "fast");
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange follower = request();
+
+ StepVerifier.create(execute(follower, chain)).verifyComplete();
+
+ assertEquals(2, calls.get());
+ assertEquals("fast",
follower.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ @Test
+ void testDistinctKeysFillIndependently() throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class, new
ResponseSingleFlight(64, 64, 5000L, 30000L, 1048576L));
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ ShenyuPluginChain chain = exchange -> {
+ calls.incrementAndGet();
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ leaderStarted.countDown();
+ return gate.asMono().then(write(exchange, "body"));
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange other = MockServerWebExchange.from(
+ MockServerHttpRequest.get(URI + "/other").build());
+
+ StepVerifier.create(execute(other, chain))
+ .then(() -> gate.tryEmitEmpty())
+ .verifyComplete();
+
+ assertEquals(2, calls.get());
+ assertEquals("body",
other.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ @Test
+ void testCancelledFollowerReleasesItsSlot() throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class, new
ResponseSingleFlight(64, 64, 5000L, 30000L, 1048576L));
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ ShenyuPluginChain chain = exchange -> {
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ leaderStarted.countDown();
+ return gate.asMono().then(write(exchange, "shared"));
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange cancelled = request();
+ StepVerifier.create(execute(cancelled,
chain)).thenCancel().verify();
+
+ MockServerWebExchange follower = request();
+ StepVerifier.create(execute(follower, chain))
+ .then(() -> gate.tryEmitEmpty())
+ .verifyComplete();
+
+ assertEquals("shared",
follower.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ @Test
+ void testFillsWhenNoKeySlotIsAvailable() throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class, new
ResponseSingleFlight(1, 64, 50L, 30000L, 1048576L));
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ ShenyuPluginChain chain = exchange -> {
+ int call = calls.incrementAndGet();
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ if (call == 1) {
+ leaderStarted.countDown();
+ return gate.asMono().then(write(exchange, "slow"));
+ }
+ return write(exchange, "fast-" + call);
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange otherKey = MockServerWebExchange.from(
+ MockServerHttpRequest.get(URI + "/capacity").build());
+
+ StepVerifier.create(execute(otherKey, chain)).verifyComplete();
+
+ assertEquals(2, calls.get());
+ assertEquals("fast-2",
otherKey.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ @Test
+ void testFollowerFillsWhenLeaderChainFails() throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class, new
ResponseSingleFlight(64, 64, 5000L, 30000L, 1048576L));
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ IllegalStateException failure = new IllegalStateException("upstream
failed");
+ ShenyuPluginChain chain = exchange -> {
+ int call = calls.incrementAndGet();
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ if (call == 1) {
+ leaderStarted.countDown();
+ return gate.asMono().then(Mono.error(failure));
+ }
+ return write(exchange, "recovered");
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange follower = request();
+
+ StepVerifier.create(execute(follower, chain))
+ .then(() -> gate.tryEmitEmpty())
+ .verifyComplete();
+
+ assertEquals(2, calls.get());
+ assertEquals("recovered",
follower.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {201, 404})
+ void testFollowerReplaysLeaderStatus(final int status) throws Exception {
+ Singleton.INST.single(ResponseSingleFlight.class, new
ResponseSingleFlight(64, 64, 5000L, 30000L, 1048576L));
+ Sinks.One<Void> gate = Sinks.one();
+ CountDownLatch leaderStarted = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ ShenyuPluginChain chain = exchange -> {
+ calls.incrementAndGet();
+
exchange.getResponse().setStatusCode(HttpStatusCode.valueOf(status));
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ exchange.getResponse().getHeaders().set("X-Upstream", "leader");
+ leaderStarted.countDown();
+ return gate.asMono().then(write(exchange, "body"));
+ };
+ MockServerWebExchange leader = request();
+ Disposable leaderSubscription = execute(leader, chain).subscribe();
+ try {
+ assertTrue(leaderStarted.await(5, TimeUnit.SECONDS));
+ MockServerWebExchange follower = request();
+
+ StepVerifier.create(execute(follower, chain))
+ .then(() -> gate.tryEmitEmpty())
+ .verifyComplete();
+
+ assertEquals(1, calls.get());
+ assertEquals(status,
follower.getResponse().getStatusCode().value());
+ assertEquals("leader",
follower.getResponse().getHeaders().getFirst("X-Upstream"));
+ assertEquals("body",
follower.getResponse().getBodyAsString().block());
+ } finally {
+ leaderSubscription.dispose();
+ }
+ }
+
+ private Mono<Void> execute(final ServerWebExchange exchange, final
ShenyuPluginChain chain) {
+ return plugin.doExecute(exchange, chain, selector, rule);
+ }
+
+ private static Mono<Void> write(final ServerWebExchange exchange, final
String body) {
+ return
exchange.getResponse().writeWith(Mono.just(exchange.getResponse().bufferFactory().wrap(body.getBytes(StandardCharsets.UTF_8))));
+ }
+
+ private static MockServerWebExchange request() {
+ return
MockServerWebExchange.from(MockServerHttpRequest.get(URI).build());
+ }
+}
