Release ByteBuf on channel write failure in ReactorNettyClient

[bugs84]

Problem

When a Netty channel write fails (broken connection, RST packet, connection refused), the ByteBuf produced by FrontendMessage.encode() is never released, causing a direct memory leak detectable by Netty's ResourceLeakDetector.

  io.netty.util.ResourceLeakDetector : LEAK: ByteBuf.release() was not called
    before it's garbage-collected.
    ...
    io.r2dbc.postgresql.message.frontend.Query.lambda$encode$0(Query.java:60)
    ...
    io.r2dbc.postgresql.client.ReactorNettyClient.doSendRequest(...)

Root Cause

All 21 FrontendMessage.encode() implementations use Mono.fromSupplier() with no cleanup hook. ReactorNettyClient subscribes to encode() and passes the resulting ByteBuf to connection.outbound().send(). When the channel write fails, reactor-netty does not release the ByteBuf —
and neither does ReactorNettyClient.

Fix

Separate encode from send: materialize the ByteBuf first via Flux.from(encode()), then write it with doOnError and doOnCancel hooks that release the buffer if the write fails or is cancelled.

return Flux.from(message.encode(this.byteBufAllocator))
    .flatMap(buf ->
        connection.outbound().send(Mono.just(buf)).then()
            .doOnError(e -> {
                if (buf.refCnt() > 0) {
                    ReferenceCountUtil.release(buf);
                }
            })
            .doOnCancel(() -> {
                if (buf.refCnt() > 0) {
                    ReferenceCountUtil.release(buf);
                }
            })
    )
    .then();

The refCnt() > 0 check prevents double-release — on successful write, Netty's channel pipeline already releases the buffer.

Fixed in both locations:

• The request pipeline (constructor)
• The close() method (Terminate message)

Why consumer-side fix (not producer-side)

• Fixing each of 21 encode() methods with Mono.using() would be fragile and repetitive
• Mono.using() cleanup risks double-release since Netty releases on successful write
• One fix in ReactorNettyClient covers all message types

Reproducer

Full reproducer with deterministic proof (no Docker needed) and integration tests:
https://github.com/bugs84/r2dbc-postgresql-bytebuf-leak-reproducer

References

Fixes #580
Related: #393, #248