Skip to content

Commit a2887c6

Browse files
kannanjgithubkannanjgithub
committed
xds: cap flow control window at 64-bit signed integer limits to prevent overflow
Use LongMath.saturatedAdd when processing ServerWindowUpdate and replenishing sidestream windows to prevent integer overflow from turning windows negative. TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0
1 parent 7628b0d commit a2887c6

2 files changed

Lines changed: 256 additions & 6 deletions

File tree

‎xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import com.google.common.annotations.VisibleForTesting;
2929
import com.google.common.collect.ImmutableList;
3030
import com.google.common.collect.ImmutableMap;
31+
import com.google.common.math.LongMath;
3132
import com.google.protobuf.ByteString;
3233
import com.google.protobuf.Struct;
3334
import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ProcessingMode;
@@ -495,8 +496,10 @@ public void onNext(ProcessingResponse response) {
495496
boolean wasReady;
496497
synchronized (streamLock) {
497498
wasReady = isReady();
498-
downstreamToSidestreamWindow += update.getWindowIncrementDownstreamToSidestream();
499-
upstreamToSidestreamWindow += update.getWindowIncrementUpstreamToSidestream();
499+
downstreamToSidestreamWindow = LongMath.saturatedAdd(
500+
downstreamToSidestreamWindow, update.getWindowIncrementDownstreamToSidestream());
501+
upstreamToSidestreamWindow = LongMath.saturatedAdd(
502+
upstreamToSidestreamWindow, update.getWindowIncrementUpstreamToSidestream());
500503
drainPendingRequestBodyMessages();
501504
drainPendingRequests();
502505
if (wrappedListener != null) {
@@ -779,8 +782,8 @@ void mergeAccumulatedWindowUpdates(ProcessingRequest.Builder requestBuilder) {
779782
.build());
780783
accumulatedWindowUpdateSidestreamToUpstream -= incrementUpstream;
781784
accumulatedWindowUpdateSidestreamToDownstream -= incrementDownstream;
782-
sidestreamToUpstreamWindow += incrementUpstream;
783-
sidestreamToDownstreamWindow += incrementDownstream;
785+
sidestreamToUpstreamWindow = LongMath.saturatedAdd(sidestreamToUpstreamWindow, incrementUpstream);
786+
sidestreamToDownstreamWindow = LongMath.saturatedAdd(sidestreamToDownstreamWindow, incrementDownstream);
784787
}
785788
}
786789

@@ -803,8 +806,8 @@ private void trySendAccumulatedWindowUpdates() {
803806
if (shouldSend) {
804807
accumulatedWindowUpdateSidestreamToUpstream -= incrementUpstream;
805808
accumulatedWindowUpdateSidestreamToDownstream -= incrementDownstream;
806-
sidestreamToUpstreamWindow += incrementUpstream;
807-
sidestreamToDownstreamWindow += incrementDownstream;
809+
sidestreamToUpstreamWindow = LongMath.saturatedAdd(sidestreamToUpstreamWindow, incrementUpstream);
810+
sidestreamToDownstreamWindow = LongMath.saturatedAdd(sidestreamToDownstreamWindow, incrementDownstream);
808811

809812
sendToExtProc(ProcessingRequest.newBuilder()
810813
.setClientWindowUpdate(ProcessingRequest.ClientWindowUpdate.newBuilder()

‎xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java‎

Lines changed: 247 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12710,6 +12710,253 @@ public void onCompleted() {
1271012710
channelManager.close();
1271112711
}
1271212712

12713+
@Test
12714+
@SuppressWarnings("unchecked")
12715+
public void testServerWindowUpdate_downstreamToSidestreamOverflowAvoidedAndNextMessageSent()
12716+
throws Exception {
12717+
ExternalProcessor proto = ExternalProcessor.newBuilder()
12718+
.setGrpcService(GrpcService.newBuilder()
12719+
.setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder()
12720+
.setTargetUri("in-process:///" + extProcServerName)
12721+
.addChannelCredentialsPlugin(Any.newBuilder()
12722+
.setTypeUrl(INSECURE_CREDENTIALS_TYPE_URL)
12723+
.build())
12724+
.build())
12725+
.build())
12726+
.setProcessingMode(ProcessingMode.newBuilder()
12727+
.setRequestBodyMode(ProcessingMode.BodySendMode.GRPC)
12728+
.setResponseBodyMode(ProcessingMode.BodySendMode.NONE)
12729+
.setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP)
12730+
.setResponseTrailerMode(ProcessingMode.HeaderSendMode.SKIP)
12731+
.build())
12732+
.build();
12733+
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
12734+
provider.parseFilterConfig(Any.pack(proto), filterContext);
12735+
assertThat(configOrError.errorDetail).isNull();
12736+
ExternalProcessorFilterConfig filterConfig = configOrError.config;
12737+
12738+
final List<ProcessingRequest> receivedRequests = new CopyOnWriteArrayList<>();
12739+
final CountDownLatch bodyLatch = new CountDownLatch(1);
12740+
final AtomicReference<StreamObserver<ProcessingResponse>> responseObserverRef =
12741+
new AtomicReference<>();
12742+
12743+
ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
12744+
new ExternalProcessorGrpc.ExternalProcessorImplBase() {
12745+
@Override
12746+
public StreamObserver<ProcessingRequest> process(
12747+
final StreamObserver<ProcessingResponse> responseObserver) {
12748+
responseObserverRef.set(responseObserver);
12749+
((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
12750+
return new StreamObserver<ProcessingRequest>() {
12751+
@Override
12752+
public void onNext(ProcessingRequest request) {
12753+
receivedRequests.add(request);
12754+
if (request.hasRequestHeaders()) {
12755+
responseObserver.onNext(ProcessingResponse.newBuilder()
12756+
.setRequestHeaders(HeadersResponse.newBuilder().build())
12757+
.setServerWindowUpdate(ProcessingResponse.ServerWindowUpdate.newBuilder()
12758+
.setWindowIncrementDownstreamToSidestream(Long.MAX_VALUE)
12759+
.build())
12760+
.build());
12761+
} else if (request.hasRequestBody()) {
12762+
bodyLatch.countDown();
12763+
}
12764+
}
12765+
12766+
@Override
12767+
public void onError(Throwable t) {}
12768+
12769+
@Override
12770+
public void onCompleted() {
12771+
responseObserver.onCompleted();
12772+
}
12773+
};
12774+
}
12775+
};
12776+
12777+
String uniqueExtProcServerName = InProcessServerBuilder.generateName();
12778+
grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName)
12779+
.addService(extProcImpl)
12780+
.directExecutor()
12781+
.build().start());
12782+
12783+
CachedChannelManager channelManager = new CachedChannelManager(config -> {
12784+
return grpcCleanup.register(
12785+
InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build());
12786+
});
12787+
12788+
ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor(
12789+
filterConfig, channelManager, scheduler, FAKE_CONTEXT);
12790+
12791+
dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService")
12792+
.addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall(
12793+
(request, responseObserver) -> {
12794+
responseObserver.onNext("Hello " + request);
12795+
responseObserver.onCompleted();
12796+
}))
12797+
.build());
12798+
12799+
ManagedChannel dataPlaneChannel = grpcCleanup.register(
12800+
InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build());
12801+
12802+
CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor());
12803+
ClientCall<String, String> proxyCall =
12804+
interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel);
12805+
12806+
proxyCall.start(new ClientCall.Listener<String>() {}, new Metadata());
12807+
proxyCall.request(1);
12808+
12809+
// Verify proxyCall is ready and not blocked by negative window overflow
12810+
assertThat(proxyCall.isReady()).isTrue();
12811+
12812+
proxyCall.sendMessage("hello");
12813+
12814+
// Verify request body reached ext_proc without being blocked by negative overflow
12815+
assertThat(bodyLatch.await(5, TimeUnit.SECONDS)).isTrue();
12816+
assertThat(receivedRequests.size()).isAtLeast(2);
12817+
assertThat(receivedRequests.get(1).hasRequestBody()).isTrue();
12818+
assertThat(receivedRequests.get(1).getRequestBody().getBody().toStringUtf8()).isEqualTo("hello");
12819+
12820+
proxyCall.cancel("Cleanup", null);
12821+
channelManager.close();
12822+
}
12823+
12824+
@Test
12825+
@SuppressWarnings("unchecked")
12826+
public void testServerWindowUpdate_upstreamToSidestreamOverflowAvoidedAndNextMessageSent()
12827+
throws Exception {
12828+
ExternalProcessor proto = ExternalProcessor.newBuilder()
12829+
.setGrpcService(GrpcService.newBuilder()
12830+
.setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder()
12831+
.setTargetUri("in-process:///" + extProcServerName)
12832+
.addChannelCredentialsPlugin(Any.newBuilder()
12833+
.setTypeUrl(INSECURE_CREDENTIALS_TYPE_URL)
12834+
.build())
12835+
.build())
12836+
.build())
12837+
.setProcessingMode(ProcessingMode.newBuilder()
12838+
.setRequestBodyMode(ProcessingMode.BodySendMode.NONE)
12839+
.setResponseBodyMode(ProcessingMode.BodySendMode.GRPC)
12840+
.setRequestHeaderMode(ProcessingMode.HeaderSendMode.SKIP)
12841+
.setResponseHeaderMode(ProcessingMode.HeaderSendMode.SEND)
12842+
.setResponseTrailerMode(ProcessingMode.HeaderSendMode.SEND)
12843+
.build())
12844+
.build();
12845+
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
12846+
provider.parseFilterConfig(Any.pack(proto), filterContext);
12847+
assertThat(configOrError.errorDetail).isNull();
12848+
ExternalProcessorFilterConfig filterConfig = configOrError.config;
12849+
12850+
final List<ProcessingRequest> receivedRequests = new CopyOnWriteArrayList<>();
12851+
final CountDownLatch responseBodyLatch = new CountDownLatch(1);
12852+
final AtomicReference<StreamObserver<ProcessingResponse>> responseObserverRef =
12853+
new AtomicReference<>();
12854+
12855+
ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
12856+
new ExternalProcessorGrpc.ExternalProcessorImplBase() {
12857+
@Override
12858+
public StreamObserver<ProcessingRequest> process(
12859+
final StreamObserver<ProcessingResponse> responseObserver) {
12860+
responseObserverRef.set(responseObserver);
12861+
((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
12862+
return new StreamObserver<ProcessingRequest>() {
12863+
@Override
12864+
public void onNext(ProcessingRequest request) {
12865+
receivedRequests.add(request);
12866+
if (request.hasResponseHeaders()) {
12867+
// Send ServerWindowUpdate with Long.MAX_VALUE increment for upstreamToSidestream
12868+
responseObserver.onNext(ProcessingResponse.newBuilder()
12869+
.setResponseHeaders(HeadersResponse.newBuilder().build())
12870+
.setServerWindowUpdate(ProcessingResponse.ServerWindowUpdate.newBuilder()
12871+
.setWindowIncrementUpstreamToSidestream(Long.MAX_VALUE)
12872+
.build())
12873+
.build());
12874+
} else if (request.hasResponseBody()) {
12875+
responseBodyLatch.countDown();
12876+
}
12877+
}
12878+
12879+
@Override
12880+
public void onError(Throwable t) {}
12881+
12882+
@Override
12883+
public void onCompleted() {
12884+
responseObserver.onCompleted();
12885+
}
12886+
};
12887+
}
12888+
};
12889+
12890+
String uniqueExtProcServerName = InProcessServerBuilder.generateName();
12891+
grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName)
12892+
.addService(extProcImpl)
12893+
.directExecutor()
12894+
.build().start());
12895+
12896+
CachedChannelManager channelManager = new CachedChannelManager(config -> {
12897+
return grpcCleanup.register(
12898+
InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build());
12899+
});
12900+
12901+
ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor(
12902+
filterConfig, channelManager, scheduler, FAKE_CONTEXT);
12903+
12904+
final AtomicReference<StreamObserver<String>> dataPlaneResponseObserverRef =
12905+
new AtomicReference<>();
12906+
final CountDownLatch upstreamConnectedLatch = new CountDownLatch(1);
12907+
dataPlaneServiceRegistry.addService(
12908+
ServerServiceDefinition.builder("test.TestService")
12909+
.addMethod(
12910+
METHOD_BIDI_STREAMING,
12911+
ServerCalls.asyncBidiStreamingCall(
12912+
new ServerCalls.BidiStreamingMethod<String, String>() {
12913+
@Override
12914+
public StreamObserver<String> invoke(
12915+
StreamObserver<String> responseObserver) {
12916+
dataPlaneResponseObserverRef.set(responseObserver);
12917+
upstreamConnectedLatch.countDown();
12918+
return new StreamObserver<String>() {
12919+
@Override
12920+
public void onNext(String value) {}
12921+
12922+
@Override
12923+
public void onError(Throwable t) {}
12924+
12925+
@Override
12926+
public void onCompleted() {}
12927+
};
12928+
}
12929+
}))
12930+
.build());
12931+
12932+
ManagedChannel dataPlaneChannel = grpcCleanup.register(
12933+
InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build());
12934+
12935+
ClientCall<String, String> proxyCall =
12936+
interceptCall(interceptor, METHOD_BIDI_STREAMING,
12937+
DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()),
12938+
dataPlaneChannel);
12939+
12940+
proxyCall.start(new ClientCall.Listener<String>() {}, new Metadata());
12941+
proxyCall.request(10);
12942+
12943+
assertThat(upstreamConnectedLatch.await(5, TimeUnit.SECONDS)).isTrue();
12944+
StreamObserver<String> upstreamResponseObserver = dataPlaneResponseObserverRef.get();
12945+
assertThat(upstreamResponseObserver).isNotNull();
12946+
12947+
// Upstream sends response message. Without saturated add, window would wrap to negative
12948+
// and buffer the message. With saturated add, it is sent to ext_proc immediately.
12949+
upstreamResponseObserver.onNext("server message");
12950+
12951+
assertThat(responseBodyLatch.await(5, TimeUnit.SECONDS)).isTrue();
12952+
assertThat(receivedRequests.size()).isAtLeast(2);
12953+
assertThat(receivedRequests.get(1).hasResponseBody()).isTrue();
12954+
assertThat(receivedRequests.get(1).getResponseBody().getBody().toStringUtf8()).isEqualTo("server message");
12955+
12956+
proxyCall.cancel("Cleanup", null);
12957+
channelManager.close();
12958+
}
12959+
1271312960
@Test
1271412961
@SuppressWarnings("unchecked")
1271512962
public void testDownstreamToSidestreamFlowControl_EnforcesWindow() throws Exception {

0 commit comments

Comments
 (0)