From 1df47dbaa7dd6599b7b43986eaf43556c2ba5ef4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20Quenaudon?= Date: Fri, 28 Aug 2026 15:27:40 +0100 Subject: [PATCH] Enforce user-set gRPC call timeouts on the client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A timeout or a deadline set on GrpcCall.timeout or GrpcStreamingCall.timeout was serialized into the grpc-timeout request header, but the client never enforced it. When the server or a proxy ignores that header, or when the connection stalls, nothing bounded the call on the client. grpc-java sends the header and also runs a client-side deadline timer. Wire only did the header half. Now initCall copies the user-set timeout and deadline onto the OkHttp per-call timeout before the call starts. OkHttp cancels the call when the bound is reached. Behavior is unchanged when the user sets no value. The header logic is unchanged. Co-authored-by: Benoît Quenaudon Signed-off-by: Benoît Quenaudon --- .../squareup/wire/internal/RealGrpcCall.kt | 14 +- .../wire/internal/RealGrpcStreamingCall.kt | 12 +- .../squareup/wire/GrpcClientTimeoutTest.kt | 169 ++++++++++++++++++ 3 files changed, 190 insertions(+), 5 deletions(-) create mode 100644 wire-grpc-tests/src/test/java/com/squareup/wire/GrpcClientTimeoutTest.kt diff --git a/wire-grpc-client/src/jvmMain/kotlin/com/squareup/wire/internal/RealGrpcCall.kt b/wire-grpc-client/src/jvmMain/kotlin/com/squareup/wire/internal/RealGrpcCall.kt index 19a6e7a5d1..c9b88f1e67 100644 --- a/wire-grpc-client/src/jvmMain/kotlin/com/squareup/wire/internal/RealGrpcCall.kt +++ b/wire-grpc-client/src/jvmMain/kotlin/com/squareup/wire/internal/RealGrpcCall.kt @@ -144,11 +144,17 @@ internal class RealGrpcCall( val result = grpcClient.newCall(method, requestMetadata, requestBody, timeout) this.call = result if (canceled) result.cancel() - // If the timeout doesn't have a deadline or timeout, then the user - // didn't set the timeout on this Call manually. - if (!timeout.hasDeadline() && (timeout.timeoutNanos() == 0L)) { - (timeout as ForwardingTimeout).setDelegate(result.timeout()) + // Copy the user-set timeout and deadline onto the OkHttp timeout, which enforces them by + // canceling the call. The grpc-timeout header alone is not enforced if the server or a proxy + // ignores it. + val okHttpTimeout = result.timeout() + if (timeout.timeoutNanos() != 0L) { + okHttpTimeout.timeout(timeout.timeoutNanos(), TimeUnit.NANOSECONDS) } + if (timeout.hasDeadline()) { + okHttpTimeout.deadlineNanoTime(timeout.deadlineNanoTime()) + } + (timeout as ForwardingTimeout).setDelegate(okHttpTimeout) return result } } diff --git a/wire-grpc-client/src/jvmMain/kotlin/com/squareup/wire/internal/RealGrpcStreamingCall.kt b/wire-grpc-client/src/jvmMain/kotlin/com/squareup/wire/internal/RealGrpcStreamingCall.kt index ea7ca3b504..5465438497 100644 --- a/wire-grpc-client/src/jvmMain/kotlin/com/squareup/wire/internal/RealGrpcStreamingCall.kt +++ b/wire-grpc-client/src/jvmMain/kotlin/com/squareup/wire/internal/RealGrpcStreamingCall.kt @@ -131,7 +131,17 @@ internal class RealGrpcStreamingCall( val result = grpcClient.newCall(method, requestMetadata, requestBody, timeout) this.call = result if (canceled) result.cancel() - (timeout as ForwardingTimeout).setDelegate(result.timeout()) + // Copy the user-set timeout and deadline onto the OkHttp timeout, which enforces them by + // canceling the call. The grpc-timeout header alone is not enforced if the server or a proxy + // ignores it. + val okHttpTimeout = result.timeout() + if (timeout.timeoutNanos() != 0L) { + okHttpTimeout.timeout(timeout.timeoutNanos(), TimeUnit.NANOSECONDS) + } + if (timeout.hasDeadline()) { + okHttpTimeout.deadlineNanoTime(timeout.deadlineNanoTime()) + } + (timeout as ForwardingTimeout).setDelegate(okHttpTimeout) return result } } diff --git a/wire-grpc-tests/src/test/java/com/squareup/wire/GrpcClientTimeoutTest.kt b/wire-grpc-tests/src/test/java/com/squareup/wire/GrpcClientTimeoutTest.kt new file mode 100644 index 0000000000..356fa5bc35 --- /dev/null +++ b/wire-grpc-tests/src/test/java/com/squareup/wire/GrpcClientTimeoutTest.kt @@ -0,0 +1,169 @@ +/* + * Copyright (C) 2026 Square, Inc. + * + * Licensed 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 + * + * https://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 com.squareup.wire + +import assertk.assertFailure +import assertk.assertThat +import assertk.assertions.isInstanceOf +import assertk.assertions.isLessThan +import assertk.assertions.isTrue +import java.time.Duration +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import okhttp3.OkHttpClient +import okhttp3.Protocol +import okhttp3.mockwebserver.Dispatcher +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import okhttp3.mockwebserver.RecordedRequest +import okhttp3.mockwebserver.SocketPolicy +import okio.IOException +import org.junit.After +import org.junit.Before +import org.junit.Rule +import org.junit.Test +import org.junit.rules.Timeout +import routeguide.Point +import routeguide.RouteGuideClient + +/** + * Proves that a timeout or a deadline set on a call's [GrpcCall.timeout] is enforced on the + * client. The server in these tests receives each request and never responds. It ignores the + * `grpc-timeout` request header, like any server or proxy that does not implement gRPC deadlines. + * Only client-side enforcement can bound these calls. + * + * The OkHttpClient has no call timeout, and its read and write timeouts are far above the bounds + * that these tests assert. When a call fails fast, the per-call timeout did it. + */ +class GrpcClientTimeoutTest { + @JvmField @Rule + val mockWebServer = MockWebServer() + + @JvmField @Rule + val testTimeout = Timeout(30, TimeUnit.SECONDS) + + private lateinit var routeGuideService: RouteGuideClient + + /** Held until the test ends. The dispatcher blocks on it, so responses never go out. */ + private val dispatcherRelease = CountDownLatch(1) + + @Before + fun setUp() { + mockWebServer.dispatcher = object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + dispatcherRelease.await(1, TimeUnit.MINUTES) + return MockResponse().apply { socketPolicy = SocketPolicy.NO_RESPONSE } + } + } + mockWebServer.protocols = listOf(Protocol.H2_PRIOR_KNOWLEDGE) + + val okhttpClient = OkHttpClient.Builder() + .protocols(listOf(Protocol.H2_PRIOR_KNOWLEDGE)) + .readTimeout(Duration.ofSeconds(10)) + .writeTimeout(Duration.ofSeconds(10)) + .build() + val grpcClient = GrpcClient.Builder() + .client(okhttpClient) + .baseUrl(mockWebServer.url("/")) + .build() + routeGuideService = grpcClient.create(RouteGuideClient::class) + } + + @After + fun tearDown() { + dispatcherRelease.countDown() + } + + @Test + fun unaryCallTimeoutIsEnforcedOnClient() { + val grpcCall = routeGuideService.GetFeature() + grpcCall.timeout.timeout(500, TimeUnit.MILLISECONDS) + + val elapsedMillis = elapsedMillis { + assertFailure { + grpcCall.executeBlocking(Point(latitude = 5, longitude = 6)) + }.isInstanceOf() + } + + assertThat(grpcCall.isCanceled()).isTrue() + assertThat(elapsedMillis).isLessThan(5_000L) + } + + @Test + fun unaryCallDeadlineIsEnforcedOnClient() { + val grpcCall = routeGuideService.GetFeature() + grpcCall.timeout.deadline(500, TimeUnit.MILLISECONDS) + + val elapsedMillis = elapsedMillis { + assertFailure { + grpcCall.executeBlocking(Point(latitude = 5, longitude = 6)) + }.isInstanceOf() + } + + assertThat(grpcCall.isCanceled()).isTrue() + assertThat(elapsedMillis).isLessThan(5_000L) + } + + @Test + fun streamingCallTimeoutIsEnforcedOnClient() { + val grpcCall = routeGuideService.RouteChat() + grpcCall.timeout.timeout(500, TimeUnit.MILLISECONDS) + + val (requestSink, responseSource) = grpcCall.executeBlocking() + val elapsedMillis = elapsedMillis { + assertFailure { + responseSource.read() + }.isInstanceOf() + } + + assertThat(grpcCall.isCanceled()).isTrue() + assertThat(elapsedMillis).isLessThan(5_000L) + + try { + requestSink.close() + } catch (_: IOException) { + // Closing the request stream of a canceled call may fail. That is fine here. + } + } + + @Test + fun streamingCallDeadlineIsEnforcedOnClient() { + val grpcCall = routeGuideService.RouteChat() + grpcCall.timeout.deadline(500, TimeUnit.MILLISECONDS) + + val (requestSink, responseSource) = grpcCall.executeBlocking() + val elapsedMillis = elapsedMillis { + assertFailure { + responseSource.read() + }.isInstanceOf() + } + + assertThat(grpcCall.isCanceled()).isTrue() + assertThat(elapsedMillis).isLessThan(5_000L) + + try { + requestSink.close() + } catch (_: IOException) { + // Closing the request stream of a canceled call may fail. That is fine here. + } + } + + private inline fun elapsedMillis(block: () -> Unit): Long { + val startNanos = System.nanoTime() + block() + return (System.nanoTime() - startNanos) / 1_000_000L + } +}