Implement stream operations in the conformance client #196

jhump · 2024-01-12T17:38:01Z

I've got each stream type in its own commit, to hopefully make this easier to review.

We still don't run the conformance tests for stream operations in CI because there are some flaky failures that I need to troubleshoot. In some circumstances, the reference servers are sending back unexpected errors for server stream RPCs, so I'll need to use a debugger with the reference server Go code to figure it out.

I want to dig a little bit more into the timeout issues with bidi RPCs, but I think this is a real bug in okhttp and, if so, will file an issue.

…ize'

…server impl + gzip...

…ing though(?)

… for any bidi 🤔

jhump · 2024-01-12T17:39:45Z

conformance/client/lite-unary-config.yaml

-  supportsTlsClientCerts: true
-  supportsHalfDuplexBidiOverHttp1: true


I've removed the line about client certs since this property defaults to false. An error is occurring when this code tries to instantiate a client cert from the data in the conformance request. I haven't dug into it, and decided to just omit it for now.

It turns out that we actually can't do half-duplex stuff over HTTP 1.1, because we don't actually know ahead of time, for a given bidi RPC, if it's half-duplex or full-duplex (in the framework, based on generated protobuf metadata). But the request body must be marked as "duplex" if it might be full-duplex. So have to mark the body as duplex for all bidi RPCs, and okhttp then disallows them to be used with HTTP 1.1.

Do we want to comment these properties out with a TODO to know to come back to them instead of removing them? Up to you but might help identify remaining work since it sounds like there are several things to track down.

Well, I think we can leave the supportsHalfDuplexBidiOverHttp1 out since there's no realistic way of changing that. But, yes, I should add back supportsTlsClientCerts with a TODO to get them working.

jhump · 2024-01-12T17:41:49Z

conformance/client/src/main/kotlin/com/connectrpc/conformance/client/adapt/BidiStreamClient.kt

+        val requests: RequestStream<Req>
+        val responses: ResponseStream<Resp>


I've changed a few things in these interfaces to be property getters instead of methods since they really are lightweight properties.

jhump · 2024-01-12T17:45:19Z

...ormance/client/src/main/kotlin/com/connectrpc/conformance/client/adapt/ClientStreamClient.kt

@@ -46,6 +46,7 @@ abstract class ClientStreamClient<Req : MessageLite, Resp : MessageLite>(
    interface ClientStream<Req : MessageLite, Resp : MessageLite> {
        suspend fun send(req: Req)
        suspend fun closeAndReceive(): ResponseMessage<Resp>
+        suspend fun cancel()


I had to add this in order to expose the ability to cancel a client-stream RPC. I could have also called it close for consistency with other methods. What do you think? Should I change it?

TBH, I kind of don't like that the methods that cancel the RPC are named like "receive close" / "close response", since that's not very clear as to what they're really doing. Maybe when overhauling these interfaces (more-or-less making them look like the interfaces in the conformance client's adapt package), we can revisit some of the names and name this operation cancel instead of close 🤷.

I think this is fine for now - maybe there's a better API we could put together that would hook into coroutine cancelation instead of having to expose separate methods for this.

I think this is already hooked into coroutine cancellation:

unary

streaming

But, I'm not sure I 100% understand exactly how that actually works. I don't think that alone does what we want it to do: the continuation in question for streaming simply returns the bidirectional stream object. So I think the only thing that this would cancel is the actual call to the streaming RPC method. Once it returns the stream object, that continuation is complete and the cancellation no longer does anything.

So I guess we'd have to do something similar in the app code, to tie its coroutine's cancellation to cancelling the stream (which I think means we still need a cancel method of some sort). I think it would be challenging to somehow have the framework automatically manage coroutine lifetimes and tie them to the stream since ownership is unclear -- i.e. it's possible that one coroutine could pass the stream to another that runs on a different thread/worker, in which case it's unclear what coroutine cancellation semantics should be.

jhump · 2024-01-12T17:46:06Z

library/src/main/kotlin/com/connectrpc/Code.kt

@@ -71,7 +74,8 @@ enum class Code(val codeName: String, val value: Int) {
            if (value == null) {
                return UNKNOWN
            }
-            return values().first { code -> code.value == value }


This would throw if the given value was not known, which seemed wrong.

jhump · 2024-01-12T17:47:33Z

okhttp/src/main/kotlin/com/connectrpc/okhttp/OkHttpStream.kt

@@ -175,7 +172,8 @@ private class ResponseCallback(
    }
 }

-internal class PipeDuplexRequestBody(
+internal class PipeRequestBody(
+    private val duplex: Boolean,


This was necessary in order to do client or server streaming over HTTP 1.1.

Nice! I expected this might be a bit more tricky - especially for client streaming calls.

pkwarren · 2024-01-12T19:30:42Z

conformance/client/lite-unary-config.yaml

-  supportsTlsClientCerts: true
-  supportsHalfDuplexBidiOverHttp1: true


Do we want to comment these properties out with a TODO to know to come back to them instead of removing them? Up to you but might help identify remaining work since it sounds like there are several things to track down.

pkwarren · 2024-01-12T19:34:09Z

...ormance/client/src/main/kotlin/com/connectrpc/conformance/client/adapt/ClientStreamClient.kt

@@ -46,6 +46,7 @@ abstract class ClientStreamClient<Req : MessageLite, Resp : MessageLite>(
    interface ClientStream<Req : MessageLite, Resp : MessageLite> {
        suspend fun send(req: Req)
        suspend fun closeAndReceive(): ResponseMessage<Resp>
+        suspend fun cancel()


I think this is fine for now - maybe there's a better API we could put together that would hook into coroutine cancelation instead of having to expose separate methods for this.

pkwarren · 2024-01-12T19:42:02Z

okhttp/src/main/kotlin/com/connectrpc/okhttp/OkHttpStream.kt

@@ -175,7 +172,8 @@ private class ResponseCallback(
    }
 }

-internal class PipeDuplexRequestBody(
+internal class PipeRequestBody(
+    private val duplex: Boolean,


Nice! I expected this might be a bit more tricky - especially for client streaming calls.

This includes a fix to the recently observed flakiness in client streams, which was introduced (in #196). The problems are described below. The fixes are each in their own commit, so reviewing commit-by-commit might make this PR easier to read. 1. There was an issue in the use of `Result<Unit>` as the return type for `send` operations. Since the return type (`Unit`) is really a void/no-return type, calling code was never checking the result. That means that when the operation failed, it was never noticed. I'm a little surprised that we don't have linters or warnings for calls to functions that return `Result` where that return value is ignored. It turns out that this was not just a problem in my calling code, failing to check the return value for failure, but in the framework itself: the stream wrapper in `ProtocolClient` (wrapping underlying stream returned by `ConnectOkHttpClient`) was using an `onSend` callback that called the underlying stream's `send`. But the `onSend` callback simply returned `Unit` instead of `Result<Unit>`, and the method that propagated the result wasn't checking the result and throwing. I think this is the biggest commit here, and it's because I did some overhauling of `Stream`. For one, I changed it to an interface -- mainly so that we could apply a decorator pattern to it and `HTTPClientInterface` (more on that in a later PR). This makes the wrapper in `ProtocolClient` simpler -- instead of it being a full implementation, with its own atomic booleans to guard/track the close operations, it just delegates to the underlying implementation. 2. The Connect unary protocol can return a 408 status code for "canceled" and "deadline exceeded" RPC errors. But okhttp auto-retries this status code, even though the requests are not idempotent (i.e. even for POST calls). This isn't an issue with the stream conformance tests, but was noticed later after I added an extra check to the reference server so that it catches cases where a client sends an RPC for the same test case more than once. This commit adds a network interceptor to the `OkHttpClient` that will identify 408 responses that look like Connect unary responses and change their status code to 499. That is the only way I could find to prevent the retries. 3. The recently introduced flakiness in client streams is actually a rather severe issue. It was mainly observed in the new conformance suite with server streams when gzip was used, because it was all due to race conditions and the gzip operations would slow down the producer thread just enough to tickle the issue. The problem is that the `RequestBody.writeTo` function should not return before the request body is finished when the request body is not duplex. But it was calling `pipe.fold` and then immediately returning. The `fold` method swaps in a new sink in place of the read-side of the pipe and then returns, without waiting for the pipe's write side to complete. So now we use a `CountDownLatch` to wait until the writer is complete (which is signaled via a call to `close`). 4. The last issue I encountered was much less frequent, and also turned out to be a race condition. It was caused by a concurrency bug in `okio.Pipe` (square/okio#1412). Basically, some duplex operations (i.e. bidi RPCs) would infrequently timeout because, even though the stream writer had closed the pipe, the HTTP request body incorrectly remained open. I've opened a PR with a fix in the `okio` library, but I've also added a work-around for now in the code here, by using extra synchronization between the calls to `write`, `close`, and `fold`.

jhump added 5 commits January 12, 2024 12:34

get client stream operations working -- all but 'client max message s…

9f82a3d

…ize'

get server stream operations working - some strange errors with grpc …

0df45eb

…server impl + gzip...

get half-duplex bidi stream operations working - timeouts aren't work…

798285b

…ing though(?)

get full-duplex bidi stream operations working - timeouts not working…

3758032

… for any bidi 🤔

ready to merge to main

214c1c5

jhump commented Jan 12, 2024

View reviewed changes

spotless apply

ea06627

jhump requested a review from pkwarren January 12, 2024 17:49

pkwarren approved these changes Jan 12, 2024

View reviewed changes

add TODO about TLS client certs

ca1b463

jhump merged commit 30c7c69 into main Jan 16, 2024
7 checks passed

jhump deleted the jh/stream-conformance branch January 16, 2024 14:50

jhump mentioned this pull request Jan 30, 2024

Fix problems in streams that were identified by conformance tests #210

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement stream operations in the conformance client #196

Implement stream operations in the conformance client #196

jhump commented Jan 12, 2024

jhump Jan 12, 2024

pkwarren Jan 12, 2024

jhump Jan 16, 2024

jhump Jan 12, 2024

jhump Jan 12, 2024

pkwarren Jan 12, 2024

jhump Jan 16, 2024

jhump Jan 12, 2024

jhump Jan 12, 2024

pkwarren Jan 12, 2024

pkwarren Jan 12, 2024

pkwarren Jan 12, 2024

pkwarren Jan 12, 2024

		supportsTlsClientCerts: true
		supportsHalfDuplexBidiOverHttp1: true

		val requests: RequestStream<Req>
		val responses: ResponseStream<Resp>

Implement stream operations in the conformance client #196

Implement stream operations in the conformance client #196

Conversation

jhump commented Jan 12, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment