[R2DBC] Cancelled reactive queries leak HTTP connections and exhaust the connection pool

## Description

Cancelling HTTP requests in a Spring WebFlux application while a ClickHouse R2DBC query is still running can leave the underlying HTTP response unclosed.

After enough cancelled requests, HTTP connections remain leased in the Apache HttpClient connection pool. Once the pool is exhausted, subsequent ClickHouse queries cannot acquire a connection and eventually fail with `ConnectionRequestTimeoutException`.

A complete minimal reproduction project is available here:

https://github.com/tomaszzason/clickhouse

The repository contains:

* `docker-compose.yml` — starts a local ClickHouse server;
* `requests.sh` — sends 1,000 HTTP requests and cancels them almost immediately;
* `single_request.http` — sends one normal request;
* `stack_trace.log` — thread dump captured after the application stops responding;
* a minimal Spring Boot WebFlux controller using `DatabaseClient` and `clickhouse-r2dbc`.

### Root cause

The R2DBC statement adapts `ClickHouseRequest.execute()` through a Reactor `Mono` backed by a `CompletableFuture`.

The problematic lifecycle is:

```text
WebFlux request starts
→ ClickHouseStatement starts ClickHouseRequest.execute()
→ the HTTP client cancels the WebFlux request
→ the Reactor subscription is cancelled
→ the client-v1 operation continues on a ClickHouseWorker thread
→ ClickHouseStreamResponse is created after cancellation
→ ClickHouseResult is never created
→ nobody closes the late response
→ the Apache HTTP connection remains leased
```

Cancelling the `CompletableFuture` does not necessarily stop the blocking HTTP operation running in client-v1.

If the operation finishes after the Reactor subscriber has already cancelled, its `ClickHouseResponse` may no longer have an owner responsible for closing it.

There is also a second cleanup gap in `ClickHouseResult`. The response is currently closed on normal completion using:

```java
.doOnComplete(response::close)
```

This does not cover cancellation or error signals.

As a result, responses may remain open when cancellation happens either:

1. before `ClickHouseResult` is created; or
2. while the result is being consumed.

### Steps to reproduce

1. Clone the reproduction repository:

```bash
git clone https://github.com/tomaszzason/clickhouse.git
cd clickhouse
```

2. Start ClickHouse:

```bash
docker compose up -d
```

3. Start the Spring Boot application from the repository.

4. Verify that one normal request works by running the request from:

```text
single_request.http
```

5. Run the cancellation script:

```bash
./requests.sh
```

The script starts 1,000 HTTP requests and cancels them almost immediately.

6. Wait for the script to finish.

7. Run the request from `single_request.http` again.

8. Observe that the application no longer responds normally.

9. After the configured connection request timeout expires, the application reports `ConnectionRequestTimeoutException`.

10. Inspect:

```text
stack_trace.log
```

The thread dump shows `ClickHouseWorker-*` threads waiting while Apache HttpClient tries to acquire an endpoint from the exhausted connection pool.

### Error Log or Exception StackTrace

```text
org.apache.hc.core5.http.ConnectionRequestTimeoutException:
Timeout deadline: 180000 MILLISECONDS, actual: 180001 MILLISECONDS
at org.apache.hc.client5.http.impl.classic.InternalExecRuntime.acquireEndpoint(InternalExecRuntime.java:122)
at org.apache.hc.client5.http.impl.classic.ConnectExec.execute(ConnectExec.java:127)
at org.apache.hc.client5.http.impl.classic.ExecChainElement.execute(ExecChainElement.java:51)
at org.apache.hc.client5.http.impl.classic.ProtocolExec.execute(ProtocolExec.java:192)
at org.apache.hc.client5.http.impl.classic.ExecChainElement.execute(ExecChainElement.java:51)
at org.apache.hc.client5.http.impl.classic.HttpRequestRetryExec.execute(HttpRequestRetryExec.java:112)
at org.apache.hc.client5.http.impl.classic.ExecChainElement.execute(ExecChainElement.java:51)
at org.apache.hc.client5.http.impl.classic.RedirectExec.execute(RedirectExec.java:110)
at org.apache.hc.client5.http.impl.classic.ExecChainElement.execute(ExecChainElement.java:51)
at org.apache.hc.client5.http.impl.classic.InternalHttpClient.doExecute(InternalHttpClient.java:185)
at org.apache.hc.client5.http.impl.classic.CloseableHttpClient.execute(CloseableHttpClient.java:123)
at com.clickhouse.client.http.ApacheHttpConnectionImpl.post(ApacheHttpConnectionImpl.java:301)
at com.clickhouse.client.http.ClickHouseHttpClient.send(ClickHouseHttpClient.java:196)
at com.clickhouse.client.AbstractClient.sendAsync(AbstractClient.java:165)
at com.clickhouse.client.AbstractClient.lambda$execute$0(AbstractClient.java:280)
at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1789)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1090)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:614)
at java.base/java.lang.Thread.run(Thread.java:1474)
```

The complete thread dump is available in the reproduction repository:

https://github.com/tomaszzason/clickhouse/blob/master/stack_trace.log

### Expected Behaviour

Cancelling a WebFlux request must not permanently retain an HTTP connection.

If a ClickHouse request completes after the Reactor subscriber has already cancelled:

* the late `ClickHouseResponse` must be closed;
* the response stream must be closed or drained;
* the Apache HTTP endpoint must be released;
* the connection must return to the pool.

After running `requests.sh`, the request from `single_request.http` should still complete successfully without restarting the application.

Repeated request cancellation should not cause permanent connection pool exhaustion.

### Code Example

The complete executable example is available at:

https://github.com/tomaszzason/clickhouse

The controller uses the standard reactive Spring R2DBC flow:

```java
@GetMapping("/clickhouse/test")
public Mono executeQuery() {
return databaseClient.sql(QUERY)
.map((row, metadata) ->
row.get("value", Integer.class))
.one();
}
```

The cancellation scenario is generated by `requests.sh`, which follows this pattern:

```bash
for i in $(seq 1 1000); do
curl \
--silent \
--max-time 0.001 \
"http://localhost:8080/clickhouse/test" \
>/dev/null 2>&1 &
done

wait
```

The full controller, configuration and runnable script should be taken from the reproduction repository rather than duplicated in this issue.

### Suggested Fix

#### 1. Handle responses that arrive after cancellation in `ClickHouseStatement`

`ClickHouseStatement` should retain responsibility for the request until the resulting response is either:

* successfully transferred to `ClickHouseResult`; or
* closed because the downstream subscription has already been cancelled.

Conceptually:

```java
final CompletableFuture future =
request.execute();

future.whenComplete((response, throwable) -> {
if (throwable != null) {
propagateErrorIfSubscriberIsActive(throwable);
return;
}

if (subscriberWasCancelled()) {
response.close();
return;
}

emitResponse(response);
});
```

The implementation must safely handle the race between:

```text
subscriber cancellation
future completion
response emission
response ownership transfer
```

An abandoned response must be closed exactly once.

The response must not be closed after it has already been transferred to an active `ClickHouseResult`.

#### 2. Close `ClickHouseResult` for all terminal Reactor signals

Current cleanup based on:

```java
.doOnComplete(response::close)
```

only covers normal completion.

The response lifecycle should also cover cancellation and errors, for example:

```java
.doFinally(signalType -> response.close())
```

or an equivalent explicit resource-management mechanism.

Both changes are needed because cancellation can happen before or after `ClickHouseResult` is created.

#### 3. Add regression coverage

A regression test should repeatedly:

1. start a request;
2. cancel the Reactor subscription before the response is delivered;
3. allow the underlying request to finish;
4. verify that the late response is closed;
5. verify that another query can still acquire an HTTP connection.

The test should also cover cancellation while `ClickHouseResult` is already consuming a streaming response.

### Compatibility considerations

The proposed fix should not change the public R2DBC API.

Successful queries should behave exactly as before.

The user-visible change should only be that resources associated with cancelled or failed queries are released reliably.

Special care is required around cancellation races to avoid:

* closing a response still owned by an active subscriber;
* closing the same response more than once;
* losing a late response without cleanup.

### Configuration

#### Client Configuration

The complete configuration is included in:

https://github.com/tomaszzason/clickhouse

The reproduction uses:

```text
clickhouse-r2dbc 0.9.8:http
Apache HttpClient 5.6.1
connection_request_timeout = 180000 ms
```

#### Environment

* [ ] Cloud
* Client version: `clickhouse-r2dbc 0.9.8:http`
* Language version: Java 25
* OS: Windows 11 for the Spring Boot application; Docker Desktop for ClickHouse

#### ClickHouse Server

* ClickHouse Server version: See `docker-compose.yml` in the reproduction repository
* ClickHouse Server non-default settings, if any: See the repository configuration
* `CREATE TABLE` statements for tables involved: Not required
* Sample data for all these tables: Not required

The reproduction uses ClickHouse system tables and does not require project-specific schemas or data.