Release results that are discarded after cancellation

[agorbachenko]

Make sure that:

• You have read the contribution guidelines.
• Issue referenced: ReactorNettyClient stucked on cancelled Conversation if that Conversation has more than 256 rows (size of reactor.bufferSize.small) #661 (and Connection leaks (not released) when concurrent query is canceled r2dbc/r2dbc-pool#198, where this was originally reported).
• No formatting-related changes; the diff follows the surrounding style.
• Integration test included, which fails before and passes after the change.

Issue description

Resolves #661, and the connection leak reported in r2dbc/r2dbc-pool#198.

PostgresqlStatement.execute() decorates the Flux<PostgresqlResult> with Operators.discardOnCancel. After a cancellation that operator keeps the subscription and drops any further result through Operators.onDiscard. PostgresqlResult is reference-counted and drains its remaining messages in deallocate(), so a discarded result has to be released to be drained.

The doOnDiscard(ReferenceCounted.class, ReferenceCountUtil::release) that is already present is registered above the map(…), so it is not part of discardOnCancel's subscriber context and never applies to the mapped results. A result that is cancelled before it is consumed is therefore dropped without being released, and its windowUntil window is never drained.

From that point the connection is unusable:

• the Conversation stays at the head of BackendMessageSubscriber.conversations with demand == 0,
• drainLoop() leaves at the head because the head has no demand (intentional, High CPU usage when cancelled query #242),
• demandMore() cannot request from the transport while hasBufferedItems() is true,
• every conversation queued behind it starves.

State of the message subscriber once wedged, read from a running reproducer:

conversations=2 buffer=59 upstreamDemand=171 terminated=false
head conversation: demand=0 sinkCancelled=false
first buffered message=DataRow

Note sinkCancelled=false: the client never observes a cancellation, it only sees a consumer that stopped requesting. That is why the investigations in r2dbc/r2dbc-pool#198 could not pin this on reactor-pool or on usingWhen — at or below the client this is indistinguishable from a slow consumer, so no change keyed on cancellation inside ReactorNettyClient could have fixed it.

With a pool the symptom is a leaked connection: the cancelling borrower releases it, and the next borrower's query never completes even though pg_stat_activity shows the backend idle with the response fully sent.

The handler is moved rather than duplicated. Operators.discardLocalAdapter composes handlers (safeConsumer.andThen(consumer)), so registering a second one would give every operator upstream of both registrations both handlers, and the rows FluxWindowPredicate discards would be released twice. Measured over the 940 pre-existing tests in PostgresCancelIntegrationTests: 0 Error in discard hook / IllegalReferenceCountException unpatched, 80 with a second handler added, 0 with the handler moved. Since the discard context propagates upstream only, one registration below discardOnCancel covers both concerns — WindowPredicate still sees it, and now discardOnCancel does too.

New Public APIs

None.

Additional context

The new test, PostgresCancelIntegrationTests#shouldReuseConnectionWhenResultIsDiscardedAfterCancel, cancels a streaming response without consuming the result and then reuses the connection. It is deterministic — no pool and no race involved: 5/5 repetitions fail before the change (the second query times out), 5/5 pass after.

Also verified:

• full unit suite: 1408 tests, 0 failures,
• -Dtest='*IntegrationTests': 1267 tests, 0 failures, 11 skipped, including the 946 existing cancellation repetitions,
• the reproducer from Connection leaks (not released) when concurrent query is canceled r2dbc/r2dbc-pool#198 (Spring Data R2DBC, pool of 1, query zipWith a failing Mono), upgraded to Spring Boot 4.0.7 / reactor-core 3.8.6 / reactor-pool 1.2.6 / Java 25: wedges after about three queries on both 1.1.2.RELEASE and main, and survives 127 queries with no stall and no leak reports with this change,
• a racing bad-actor/good-actor harness against a real connection, based on @chemicL's r2dbc-lab with the mocks replaced by the actual driver: stalls within 15 iterations on both 1.1.2.RELEASE and main, and passes 150 iterations × 2 response sizes × 2 scenarios with this change.

Notes:

• This is not the nested-flatMap deadlock discussed at the end of ReactorNettyClient stucked on cancelled Conversation if that Conversation has more than 256 rows (size of reactor.bufferSize.small) #661. That one is a consumer that never requests; this one is a cancellation that loses the obligation to drain.
• If you prefer a different placement, FluxDiscardOnCancelSubscriber.onNext could release ReferenceCounted items it drops instead of relying on a context handler. I went with the smaller, local change.
• The other windowUntil usage, DefaultPostgresqlReplicationConnection.startReplication(), is not affected. That chain has no discardOnCancel, so a cancellation propagates to the exchange normally and the conversation terminates; and PostgresReplicationStream is not reference-counted, so there is nothing a discard handler could release. LogicalDecodeIntegrationTests passes with -Dio.netty.leakDetection.level=paranoid and no discard-hook errors.