Skip to content

Commit fe58aa2

Browse files
konluekonlue
committed
fix(divide): bind failover inflight accounting to each attempt's response lifetime (#8158)
1 parent 8046953 commit fe58aa2

4 files changed

Lines changed: 454 additions & 62 deletions

File tree

‎shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/utils/UpstreamAttemptOwner.java‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.shenyu.plugin.base.utils;
19+
20+
import org.apache.shenyu.loadbalancer.entity.Upstream;
21+
22+
import java.util.Objects;
23+
import java.util.concurrent.atomic.AtomicReference;
24+
import java.util.function.Consumer;
25+
26+
/**
27+
* Tracks which upstream attempt currently owns this request's inflight reservation.
28+
*
29+
* <p>An attempt owns a reservation exactly while it is the serving target. A failed
30+
* attempt releases only itself (identity CAS), a successful attempt hands ownership to
31+
* the final response lifetime, and a failover transfers ownership to the next target.
32+
* This keeps initial and retry attempts independently accountable (#8158).</p>
33+
*/
34+
public final class UpstreamAttemptOwner {
35+
36+
/**
37+
* Exchange attribute containing the request's attempt owner.
38+
*/
39+
public static final String ATTRIBUTE = UpstreamAttemptOwner.class.getName();
40+
41+
private final AtomicReference<Upstream> current = new AtomicReference<>();
42+
43+
private final Consumer<Upstream> releaser;
44+
45+
/**
46+
* Create an owner for one request.
47+
*
48+
* @param releaser releases one attempt's reservation
49+
*/
50+
public UpstreamAttemptOwner(final Consumer<Upstream> releaser) {
51+
this.releaser = Objects.requireNonNull(releaser);
52+
}
53+
54+
/**
55+
* Get the upstream that currently owns the reservation.
56+
*
57+
* @return the current attempt, or null when no attempt is reserved
58+
*/
59+
public Upstream current() {
60+
return current.get();
61+
}
62+
63+
/**
64+
* Transfer ownership to a new attempt. A defensive previous value is released first.
65+
*
66+
* @param upstream the new attempt
67+
*/
68+
public void move(final Upstream upstream) {
69+
if (Objects.isNull(upstream)) {
70+
return;
71+
}
72+
final Upstream previous = current.getAndSet(upstream);
73+
if (Objects.nonNull(previous) && previous != upstream) {
74+
releaser.accept(previous);
75+
}
76+
}
77+
78+
/**
79+
* Release the current attempt if one is reserved.
80+
*/
81+
public void releaseCurrent() {
82+
final Upstream serving = current.getAndSet(null);
83+
if (Objects.nonNull(serving)) {
84+
releaser.accept(serving);
85+
}
86+
}
87+
88+
/**
89+
* Release a specific attempt only while it still owns the reservation.
90+
*
91+
* @param attempt the attempt that failed or was cancelled
92+
* @return true when this call released the attempt
93+
*/
94+
public boolean release(final Upstream attempt) {
95+
if (Objects.nonNull(attempt) && current.compareAndSet(attempt, null)) {
96+
releaser.accept(attempt);
97+
return true;
98+
}
99+
return false;
100+
}
101+
}

‎shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/DefaultRetryStrategy.java‎

Lines changed: 86 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.apache.shenyu.plugin.api.utils.RequestUrlUtils;
2828
import org.apache.shenyu.plugin.base.circuitbreaker.UpstreamCircuitBreaker;
2929
import org.apache.shenyu.plugin.base.utils.LoadbalancerUtils;
30+
import org.apache.shenyu.plugin.base.utils.UpstreamAttemptOwner;
3031
import org.apache.shenyu.plugin.base.utils.UpstreamRoutingContext;
3132
import org.apache.shenyu.plugin.httpclient.exception.ShenyuTimeoutException;
3233
import org.slf4j.Logger;
@@ -35,6 +36,7 @@
3536
import org.springframework.web.server.ResponseStatusException;
3637
import org.springframework.web.server.ServerWebExchange;
3738
import reactor.core.publisher.Mono;
39+
import reactor.core.publisher.SignalType;
3840
import reactor.util.retry.Retry;
3941
import reactor.util.retry.RetryBackoffSpec;
4042

@@ -105,7 +107,19 @@ public Mono<R> execute(final Mono<R> clientResponse, final ServerWebExchange exc
105107
Mono<R> executeFailover(final Mono<R> clientResponse, final ServerWebExchange exchange, final Duration duration,
106108
final int retryTimes, final Duration backoff, final Predicate<Throwable> retryFilter) {
107109
final Set<URI> exclude = new HashSet<>(Collections.singletonList(Objects.requireNonNull(exchange.getAttribute(Constants.HTTP_URI))));
110+
final UpstreamAttemptOwner owner = exchange.getAttribute(UpstreamAttemptOwner.ATTRIBUTE);
108111
Mono<R> result = clientResponse;
112+
if (Objects.nonNull(owner)) {
113+
// Bind the release to the initial attempt itself, not to the composed failover chain:
114+
// onErrorResume below consumes the initial error, so a chain-level doFinally would only
115+
// run after the retry finished and would keep A reserved through the backoff (#8158).
116+
final Upstream initial = owner.current();
117+
result = result.doFinally(signal -> {
118+
if (signal != SignalType.ON_COMPLETE) {
119+
owner.release(initial);
120+
}
121+
});
122+
}
109123
for (int i = 0; i < retryTimes; i++) {
110124
result = result.onErrorResume(retryFilter, th -> backoff.isZero()
111125
? resend(exchange, duration, exclude)
@@ -138,33 +152,44 @@ private Mono<R> mapFailoverErrors(final Mono<R> response) {
138152
.onErrorMap(java.util.concurrent.TimeoutException.class, th -> new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, th.getMessage(), th));
139153
}
140154

141-
private Mono<R> resend(final ServerWebExchange exchange,
142-
final Duration duration,
143-
final Set<URI> exclude) {
144-
final String selectorId = exchange.getAttribute(Constants.DIVIDE_SELECTOR_ID);
145-
final String loadBalance = exchange.getAttribute(Constants.LOAD_BALANCE);
155+
private List<Upstream> retryCandidates(final ServerWebExchange exchange, final String selectorId,
156+
final Set<URI> exclude) {
146157
// Refresh health/topology without losing the request's resolved label partition.
147158
final UpstreamRoutingContext routing = exchange.getAttribute(UpstreamRoutingContext.ATTRIBUTE);
148159
final List<Upstream> availableUpstreamList = Objects.isNull(routing)
149160
? UpstreamCacheManager.getInstance().findUpstreamListBySelectorId(selectorId)
150161
: routing.filter(selectorId, UpstreamCacheManager.getInstance().findAllUpstreamListBySelectorId(selectorId));
151162
if (Objects.isNull(availableUpstreamList)) {
152-
return Mono.error(new ShenyuException("CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER"));
163+
return Collections.emptyList();
153164
}
154-
final List<Upstream> upstreamList = availableUpstreamList
155-
.stream().filter(data -> {
156-
final URI upstreamUri = URI.create(data.buildDomain());
157-
for (URI needToExclude : exclude) {
158-
if (isSameUpstream(needToExclude, upstreamUri)) {
159-
return false;
160-
}
161-
}
162-
return true;
163-
})
165+
return availableUpstreamList.stream().filter(data -> {
166+
final URI upstreamUri = URI.create(data.buildDomain());
167+
for (URI needToExclude : exclude) {
168+
if (isSameUpstream(needToExclude, upstreamUri)) {
169+
return false;
170+
}
171+
}
172+
return true;
173+
})
164174
//skip upstreams that are still blocked by the built-in circuit breaker
165175
.filter(data -> UpstreamCircuitBreaker.getInstance()
166176
.isRequestAllowed(UpstreamCircuitBreaker.buildKey(selectorId, data)))
167177
.collect(Collectors.toList());
178+
}
179+
180+
private Mono<R> resend(final ServerWebExchange exchange,
181+
final Duration duration,
182+
final Set<URI> exclude) {
183+
final UpstreamAttemptOwner owner = exchange.getAttribute(UpstreamAttemptOwner.ATTRIBUTE);
184+
if (Objects.nonNull(owner)) {
185+
// reaching a resend means the previous attempt failed; release that attempt before
186+
// selecting and reserving the next one. The release is idempotent with the failure
187+
// callback that may run just after this method starts (#8158).
188+
owner.releaseCurrent();
189+
}
190+
final String selectorId = exchange.getAttribute(Constants.DIVIDE_SELECTOR_ID);
191+
final String loadBalance = exchange.getAttribute(Constants.LOAD_BALANCE);
192+
final List<Upstream> upstreamList = retryCandidates(exchange, selectorId, exclude);
168193
if (upstreamList.isEmpty()) {
169194
// no need to retry anymore
170195
return Mono.error(new ShenyuException("CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER"));
@@ -174,27 +199,52 @@ private Mono<R> resend(final ServerWebExchange exchange,
174199
// no need to retry anymore
175200
return Mono.error(new ShenyuException("CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER"));
176201
}
177-
final URI newUri = RequestUrlUtils.buildRequestUri(exchange, upstream.buildDomain());
178-
// in order not to affect the next retry call, newUri needs to be excluded
179-
exclude.add(newUri);
180-
final String breakerKey = UpstreamCircuitBreaker.buildKey(selectorId, upstream);
181-
Mono<R> retryResponse = httpClientPlugin
182-
.doRequest(exchange, exchange.getRequest().getMethod().name(), newUri, httpClientPlugin.getCachedRequestBody(exchange))
183-
.timeout(duration, Mono.error(() -> new TimeoutException("Response took longer than timeout: " + duration)))
184-
.doOnError(e -> LOG.error(e.getMessage(), e));
185-
if (LoadBalanceEnum.P2C.getName().equals(loadBalance)) {
186-
// Selecting the retry upstream above goes through the load balancer, which increments the P2C
187-
// inflight counter of the newly picked upstream. Divide's completion callback only captures the
188-
// initially selected upstream, so release this attempt's increment to keep P2C accounting paired
189-
// across success, failure and cancellation of the failover. See #7975.
190-
retryResponse = retryResponse.doFinally(signalType -> upstream.getInflight().decrementAndGet());
191-
} else if (LoadBalanceEnum.LEAST_ACTIVITY.getName().equals(loadBalance)) {
192-
// Least active compares outstanding requests but does not reserve on select, so this attempt
193-
// stays reserved until it terminates. See #7986.
194-
upstream.getInflight().incrementAndGet();
195-
retryResponse = retryResponse.doFinally(signalType -> upstream.getInflight().decrementAndGet());
202+
final boolean ownerPresent = Objects.nonNull(owner);
203+
boolean reserved = false;
204+
try {
205+
if (LoadBalanceEnum.LEAST_ACTIVITY.getName().equals(loadBalance)) {
206+
// Least active compares outstanding requests but does not reserve on select.
207+
upstream.getInflight().incrementAndGet();
208+
reserved = true;
209+
} else if (LoadBalanceEnum.P2C.getName().equals(loadBalance)) {
210+
// P2C selection already incremented the retry target.
211+
reserved = true;
212+
}
213+
if (ownerPresent) {
214+
// Transfer ownership before constructing the request: a succeeded attempt is
215+
// released by the terminal callback when its body finishes, a failed one by the
216+
// callback below. A synchronous failure after this point releases the new target.
217+
owner.move(upstream);
218+
}
219+
final URI newUri = RequestUrlUtils.buildRequestUri(exchange, upstream.buildDomain());
220+
// in order not to affect the next retry call, newUri needs to be excluded
221+
exclude.add(newUri);
222+
final String breakerKey = UpstreamCircuitBreaker.buildKey(selectorId, upstream);
223+
Mono<R> retryResponse = httpClientPlugin
224+
.doRequest(exchange, exchange.getRequest().getMethod().name(), newUri, httpClientPlugin.getCachedRequestBody(exchange))
225+
.timeout(duration, Mono.error(() -> new TimeoutException("Response took longer than timeout: " + duration)))
226+
.doOnError(e -> LOG.error(e.getMessage(), e));
227+
if (ownerPresent) {
228+
retryResponse = retryResponse.doFinally(signal -> {
229+
if (signal != SignalType.ON_COMPLETE) {
230+
// identity CAS: an immediate resend may already have moved ownership to the
231+
// next attempt, and this callback must never release that newer target
232+
owner.release(upstream);
233+
}
234+
});
235+
} else if (reserved) {
236+
// no Divide attempt owner (non-Divide caller): keep the previous accounting
237+
retryResponse = retryResponse.doFinally(signal -> upstream.getInflight().decrementAndGet());
238+
}
239+
return UpstreamCircuitBreaker.recordOutcome(retryResponse, breakerKey);
240+
} catch (RuntimeException ex) {
241+
if (ownerPresent) {
242+
owner.release(upstream);
243+
} else if (reserved) {
244+
upstream.getInflight().decrementAndGet();
245+
}
246+
throw ex;
196247
}
197-
return UpstreamCircuitBreaker.recordOutcome(retryResponse, breakerKey);
198248
}
199249

200250
private boolean isSameUpstream(final URI first, final URI second) {

0 commit comments

Comments
 (0)