ClientOnlyStream.receiveAndClose(): Cancellation Exception occurring

[Jeremy-Brown14]

I'm getting a crash calling this function, occasionally, it doesn't occur every time but it occurs about ⅓ of the time. It seems that if the function doesn't return the resultChannel.receive() in a timely manner, it hits the finally block and cancels the resultChannel, throwing the below exception. Also, shouldn't this function close the messageStream after receiving the result? I have to call the close() function before calling the receiveAndClose() function, otherwise my client doesn't receive from the result channel and my backend service does not process the resulting stream I sent to it. Thanks for looking into this!

FATAL EXCEPTION: OkHttp Dispatcher
    PID: 20742
    java.util.concurrent.CancellationException: Channel was cancelled
        at kotlinx.coroutines.channels.BufferedChannel.cancelImpl$kotlinx_coroutines_core(BufferedChannel.kt:1765)
        at kotlinx.coroutines.channels.BufferedChannel.cancel(BufferedChannel.kt:1762)
        at kotlinx.coroutines.channels.ReceiveChannel$DefaultImpls.cancel$default(Channel.kt:297)
        at build.buf.connect.impl.ClientOnlyStream.receiveAndClose(ClientOnlyStream.kt:36)
        at build.buf.connect.impl.ClientOnlyStream$receiveAndClose$1.invokeSuspend(Unknown Source:14)
        at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
        at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
        at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:115)
        at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:100)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:584)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:793)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:697)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:684)

Here is my client code:

private suspend fun <PayloadType, RequestType, ResponseType> performNetworkCall(
        request: RequestType,
        performRequest: suspend (RequestType) -> ResponseMessage<ResponseType>,
        mapResponseOnSuccess: (ResponseMessage.Success<ResponseType>) -> PayloadType,
        enableLogs: Boolean = false
    ): Result<PayloadType> {
        if (enableLogs) {
            Logger.v("ConnectNetworkRequest: ${request.toString().redactPrivateFieldsConnectLibrary()}")
        }

        val isTokenValidResult = refreshTokenWithRetries()
        if (isTokenValidResult.isFailure) {
            return Result.failure(isTokenValidResult.exceptionOrNull() ?: Exception("Error token is no longer valid"))
        }

        return when (val response = performRequest(request)) {
            is ResponseMessage.Success -> {
                if (enableLogs) {
                    Logger.v("ConnectNetworkResponse: ${response.message.toString().redactPrivateFieldsConnectLibrary()}")
                }
                Result.success(mapResponseOnSuccess(response))
            }
            is ResponseMessage.Failure -> {
                Logger.e("ConnectNetworkResponse (Error): ${response.error}", response.error)
                Result.failure(response.error)
            }
        }
    }

override suspend fun savePhoto(
        file: File,
        imageType: String
    ): Result<String> {
        val imageInfo = ImageInfo
            .newBuilder()
            .setImageType(imageType)
            .build()

        return performNetworkCall(
            request = SavePhotoRequest
                .newBuilder()
                .setInfo(imageInfo)
                .build(),
            performRequest = { request ->
                val savePhoto = orderService.savePhoto()
                val response = savePhoto.send(request)
                val chunkData: MutableList<ByteString> = mutableListOf()
                file.forEachBlock(blockSize = AppConfig.getPhotoUploadChunkSize()) { buffer, bytesRead ->
                    Logger.v("ConnectNetworkRequest: savePhoto, bufferSize: ${buffer.size}, bytesRead: $bytesRead")
                    chunkData.add(ByteString.copyFrom(buffer, 0, bytesRead))
                }
                val response2: MutableList<Result<Unit>> = mutableListOf()
                chunkData.forEach { byteString ->
                    response2.add(
                        savePhoto.send(SavePhotoRequest
                            .newBuilder()
                            .setChunkData(byteString)
                            .build()
                        )
                    )
                }
                savePhoto.close()
                val result = savePhoto.receiveAndClose()
                if (response.isSuccess && response2.all { it.isSuccess }) {
                    ResponseMessage.Success(
                        message = result,
                        code = Code.OK,
                        headers = emptyMap(),
                        trailers = emptyMap()
                    )
                } else {
                    ResponseMessage.Failure(
                        error = ConnectError(code = Code.INTERNAL_ERROR),
                        code = Code.INTERNAL_ERROR,
                        headers = emptyMap(),
                        trailers = emptyMap()
                    )
                }
            },
            mapResponseOnSuccess = { "SUCCESS" }
        )
    }

[pkwarren]

Sorry for the delay - I'll be taking a look at this today. What type of server are you interfacing with when you see this behavior? (i.e. connect-go, grpc-go, grpc-java)

[Jeremy-Brown14]

connect-go

[Jeremy-Brown14]

I believe what is happening is that it doesn't receive the result in a timely enough manner, as the backend awaits a push notification for the photo URL, and then it hits the finally block and cancels, causing this crash, as the receive channel is still open awaiting a result. It always does process the result, so that's the upside, it's just that it crashes my application sometimes. Initially when implementing this, I could not get my server to provide the URL in its history. The result is not used immediately, it is used later on with a history request. That was when I was just calling close() and not close() then receiveAndClose().

[Jeremy-Brown14]

Awesome, fix looks great! That should absolutely take care of my issues. Thank you!

[pkwarren]

connect-kotlin v0.2.0 is now available with several fixes to streaming (including client streaming). Can you give the new release a try and let me know if it resolves the issues on your side?

[pkwarren]

Confirmed this is now working. Closing issue.