Dispose upstream connection when a cancelled exchange discards the response in NettyRoutingFilter

[raccoonback]

Problem

NettyRoutingFilter puts the upstream connection into CLIENT_RESPONSE_CONN_ATTR in the responseConnection(...) callback. The exchange can be cancelled after the response arrives but before that line runs. Then the response is discarded and nobody subscribes to it.

Two things happen in that case:

• The cleanup in NettyWriteResponseFilter already ran on doFinally(CANCEL). CLIENT_RESPONSE_CONN_ATTR was still null, so it disposed nothing.
• Nobody subscribes to connection.inbound().receive(). The upstream body stays in the FluxReceive queue.

After that, nothing releases the buffer. The two places that could have released it do not:

• FluxReceive clears the queue only in cleanQueue(), and every call to it is behind isCancelled(). There is no path for inboundDone.
• HttpClientOperations.onInboundClose() overrides ChannelOperations and does not call discardWhenNoReceiver(). So closing the channel does not help either.

Our gateway is used by mobile clients. They cancel requests often, so we get many HTTP 499. We see this leak in production every day. The number is small, but it does not stop.

LEAK: ByteBuf.release() was not called before it's garbage-collected.
Recent access records:
#1:
    Hint: ... Buffered ByteBufHolder in the inbound buffer queue
    io.netty.handler.codec.http.DefaultHttpContent.touch(...)
    reactor.netty.channel.FluxReceive.onInboundNext(...)
    reactor.netty.http.client.HttpClientOperations.onInboundNext(...)

In every report the last access is the enqueue, and there is no poll after it. So the body was never read.

Solution

Add a discard handler, so the connection is disposed when the response is discarded.

Why dispose the connection.
After the value is discarded, the connection is the only handle left to the buffered body. Disposing it cancels the inbound receiver, and cancellation is what reaches cleanQueue().
This is the same thing NettyWriteResponseFilter does in its cleanup, only from a point where the connection is still reachable.

It does not replace the existing cleanup.
Once the response reaches the filter chain, CLIENT_RESPONSE_CONN_ATTR is set and NettyWriteResponseFilter disposes the connection.
FluxTimeout, for example, drops the value with onNextDropped. That is a different hook, so the handler does not run there, but the cleanup covers it.

[raccoonback]

@ryanjbaxter @spencergibb
Hello.
I'd appreciate it if you could review this PR.

[raccoonback]

@ryanjbaxter @spencergibb
Some extra context, since this sits right next to #4278.
That one covered the case where the response is already committed. There CLIENT_RESPONSE_CONN_ATTR is set, so cleanup(exchange) has something to dispose. This PR covers the neighbouring case: the exchange is cancelled before NettyRoutingFilter stores the connection, so the same cleanup(exchange) reads a null attribute and disposes nothing.

How to occur

responseConnection(...) is _connect().flatMapMany(resp -> Flux.from(receiver.apply(resp, resp))), and both exchange.getAttributes().put(...) calls live inside receiver.

If the cancel arrives before flatMapMany subscribes to the inner publisher, receiver never runs, neither attribute is set, and resp goes out through Operators.onDiscard. resp is an HttpClientOperations, so it is a Connection. That is what doOnDiscard()` picks up here.

After that nothing releases the body. It stays in FluxReceive with no receiver, terminate() sets inboundDone, and cleanQueue() is only reachable from the cancelled path. Disposing the connection is what drains it.

In our production

Our gateway serves mobile clients, so cancelled requests are common. We added a small guard filter of our own that records the same two moments, and it has logged exactly this state. The response headers had arrived, but CLIENT_RESPONSE_CONN_ATTR was still null when the cancel came through. Disposing the connection there stopped the leak reports.

[ryanjbaxter]

LGTM. Can you submit this against the 5.0.x branch?

[raccoonback]

@ryanjbaxter
Hello!
It has been switched to the 5.0.x branch.

[raccoonback]

@ryanjbaxter
Hello!
Please check this pr.

[ryanjbaxter]

Thanks! We will merge it as soon as our release is done