connection.read() publisher may miss cancel on the server-side in case of an error

[idelpivnitskiy]

On the server-side, if service does not subscribe to the request payload body and fails the response payload body publisher, the connection will eventually be closed, but the connection.read() publisher won't be canceled nor terminated.

Therefore, NettyChannelPublisher#pending queue has 2 elements:

1. empty trailers
2. ClosedChannelException

Because there are non-consumed payload body and/or trailers, the terminal event is not delivered. IIUC the control flow the terminal signal is delayed intentionally to let the subscriber consume a successful request before terminating it.

And we miss the cancellation somewhere around SpliceFlatStreamToMetaSingle. #1105 adds a HttpTransportObserverTest#testServerFailsResponsePayloadBody that reproduces this behavior. Not critical, but we should investigate where the cancellation is missed.

Wanted but not invoked:
serverReadObserver.readCancelled();
-> at io.servicetalk.http.netty.HttpTransportObserverTest.testServerFailsResponsePayloadBody(HttpTransportObserverTest.java:310)

However, there were exactly 2 interactions with this mock:
serverReadObserver.requestedToRead(1L);
-> at io.servicetalk.transport.netty.internal.DefaultNettyConnection$2$1.lambda$request$0(DefaultNettyConnection.java:291)

serverReadObserver.itemRead();
-> at io.servicetalk.transport.netty.internal.TransportObserverUtils.safeReport(TransportObserverUtils.java:124)

See https://github.com/apple/servicetalk/pull/1105#discussion_r462529751 for more information.