Support cancelling Kotlin coroutine handlers on connection closing

[calebkiage]

I just started testing out Quarkus with Kotlin suspend handlers and it seems to work well. However, I noticed that these handlers don't support cancellation on request connection closing. I found a workaround, but it seems very cumbersome to add this to every request and would like to find out if there's a better way.

In the sample code below, endpoint /long/a cancels the delay by throwing the CancellationException when the connection is closed, but /long/b does not.

@Path("/long")
class SampleResource(private val log: Logger) {
    @GET
    @Path("a")
    @Produces(MediaType.TEXT_PLAIN)
    suspend fun longRunning(rc: RoutingContext): String = coroutineScope {
        val requestJob = Job(coroutineContext[Job])
        // Listen for the connection close event
        rc.request().connection().closeHandler {
            requestJob.cancel(CancellationException("Client connection closed"))
        }
        // Launch a coroutine within this job
        async(requestJob) {
            delay(10000)
            log.info("/long/a responding")
            "Completed request"
        }.await()
    }

    @GET
    @Path("b")
    @Produces(MediaType.TEXT_PLAIN)
    suspend fun hello(): String = coroutineScope {
        delay(10000)
        log.info("/long/b responding")
        "Completed request"
    }
}

2024-07-04 17:30:07,579 ERROR [io.qua.ver.htt.run.QuarkusErrorHandler] (vert.x-eventloop-thread-1) HTTP Request to /long/a failed, error id: 1e9c93da-0f36-484d-8557-240fba83db7f-3: java.util.concurrent.CancellationException: Client connection closed
    at ...

2024-07-04 17:30:22,693 INFO  [org.acm.SampleResource] (vert.x-eventloop-thread-0) /long/b responding

Originally posted by @calebkiage in https://github.com/quarkusio/quarkus/discussions/41696

Sample project

[quarkus-bot[bot]]

/cc @geoand (kotlin)

[geoand]

Thanks!

Mind attaching the project you used to test this?

[calebkiage]

@geoand, I've added a link to a sample project.

[geoand]

🙏🏼

[geoand]

Have a look at #41710 - it resolves your issue, but it does raise a question for us :)

[mschorsch]

Nice @geoand 👍

[geoand]

🙏

[calebkiage]

@geoand, I've tested version 3.13.0.CR1 and cancellation works for the server. I've however noticed that the client doesn't reset the connection.

When using the vert.x client manually, I can trigger cancellation by resetting the connection and it works fine:

withTimeoutOrNull(delayLong) {
    suspendCancellableCoroutine { cont->
        val options = HttpClientOptions().setDefaultHost("localhost").setDefaultPort(9000)
        val client = ctx.vertx().createHttpClient(options)
        client.request(io.vertx.core.http.HttpMethod.GET, "/long/a").onComplete { conn->
            if (conn.succeeded()) {
                cont.invokeOnCancellation {
                    // Trigger a connection reset on cancellation
                    conn.result().reset()
                }
                val request = conn.result()
                request.send().onComplete { resp->
                    if (resp.succeeded()) {
                        val body = resp.result().body()
                        body.onComplete { bodyState->
                            if (bodyState.succeeded()) {
                                cont.resume(bodyState.result().toString(Charsets.UTF_8))
                            } else {
                                cont.resumeWithException(bodyState.cause())
                            }
                        }
                    } else {
                        cont.resumeWithException(resp.cause())
                    }
                }
            } else {
                cont.resumeWithException(conn.cause())
            }
        }
    }
}

I think the generated clients should also call reset on cancellation. What do you think?