Reactor-netty: [HTTP2] bodyToMono waiting for data frame indefinitely #2

[Tiller]

Hello, I'm back again with another http2 issue :(

Expected Behavior

Return the request's body

Actual Behavior

ServerRequest.bodyToMono waits indefinitely for data even though data was sent

Steps to Reproduce

• create an HTTP2 server (H2C in my case, https://github.com/Tiller/netty-leaks simply start Application.java)
• create a controller + endpoint that takes a ServerRequest as a parameter and reads its body
• call bodyToMono on the ServerRequest
• with some http2 client, call the endpoint POST '/tap/operation' (cd client ; npm install && node test.js)

Possible Solution

Not really a possible solution, but a beginning of analysis.

When executing the following command:

echo '{"hello":"world"}' | nghttp -H':method: POST' -H'content-type: application/json' http://localhost:8081/tap/operation -d -

It works as expected. However, when running the JS client included in my github repository, which basically does the same request it fails and my bodyToMono times out

I dumped the network exchanges of both requests (see attachment) and I see one main difference:

• nghttp sends a HEADER frame, followed by a DATA frame containing the json with flag endOfStream
• the JS client sends a HEADER frame, following by a DATA frame containing the json withOUT flag endOfStream then sends another DATA frame with flag endOfStream

edit: could not attach dump, here it is: https://stuff.stooit.com/d/1/5e843f10f0f23/http2.pcap

I'd say that even if the JS exchange could be optimized, it is still a valid HTTP2 exchange and it should work as intended (note: afaik, it worked with 0.9.0.RC1)

Your Environment

OS: Xubuntu 19.10 64bits 5.3.0-42-generic
JVM: openjdk version "11.0.6" 2020-01-14
Reactor: 3.3.2.RELEASE
Spring-boot: 2.2.4.RELEASE
Reactor-netty: 0.9.4.RELEASE (tried with 0.9.6.RELEASE and same behavior is observed)
Netty: 4.1.45.Final (tried with 4.1.48.Final and the same behavior is observed)

[violetagg]

@Tiller Please upgrade Netty - https://github.com/reactor/reactor-netty/issues/965#issuecomment-584356912

[Tiller]

@violetagg thanks, but the same issue appear with netty 4.1.48.Final

[violetagg]

@Tiller I'm trying to minimise the reproducible scenario.
According to your description the server should do something like this:

@Test
public void testH2CHttp11() {
    DisposableServer server =
            HttpServer.create()
                      .protocol(HttpProtocol.H2C, HttpProtocol.HTTP11)
                      .port(8081)
                      .handle((req, res) -> res.sendString(req.receive().aggregate().asString()))
                      .wiretap(true)
                      .bindNow();

    server.onDispose()
          .block();
}

If yes I cannot reproduce your issue. Can you try it also?

[violetagg]

This one is also working

@Test
public void testHttp1or2() throws Exception {
    DisposableServer server =
            HttpServer.create()
                      .protocol(HttpProtocol.H2C, HttpProtocol.HTTP11)
                      .port(8080)
                      .handle((req, res) -> req.receive()
                                               .aggregate()
                                               .asString()
                                               .flatMap(s -> res.sendString(Mono.just(s)).then()))
                      .wiretap(true)
                      .bindNow();

    server.onDispose()
          .block();
}

[violetagg]

If you switch to the code below, what is the result?

    public Mono<ServerResponse> operationPost(final ServerRequest request) {
        return ServerResponse.ok().bodyValue(List.of(request.bodyToMono(Map.class).timeout(Duration.ofSeconds(5l)).doOnNext(System.err::println)));
    }

[Tiller]

I am sorry I am trying to write a shorter example, but I'm not fluent enough with netty/reactor-netty/spring reactor to manage to write it =X I'm still trying though.

Indeed, both your example worked as expected. And if I try to replace my method operationPost with yours, my client receive a response, but with strange data

RESPONSE [Object: null prototype] {
  ':status': 200,
  'content-type': 'application/json',
  'content-length': '24',
  'cache-control': 'no-cache, no-store, max-age=0, must-revalidate',
  pragma: 'no-cache',
  expires: '0',
  'x-content-type-options': 'nosniff',
  'x-frame-options': 'DENY',
  'x-xss-protection': '1 ; mode=block',
  'referrer-policy': 'no-referrer' }
DATA [ { scanAvailable: true } ]

(nghttp call confirms the output data)

and the printerr printed nothing in the console

edit: well, nothing is printed because you gave the list of publisher as direct body value! So no subscription :)

[Tiller]

Hmm, switching to:

    public Mono<ServerResponse> operationPost(final ServerRequest request) {
        return ServerResponse.ok().body(request.bodyToMono(Map.class).timeout(Duration.ofSeconds(5l)).doOnNext(System.err::println), Object.class);
    }

produces the same result. Ok with nghttp2, timeout with the nodejs client

[violetagg]

edit: well, nothing is printed because your gave the list of publisher as direct body value! So no subscription :)

That's happening when you write directly in the issue and not the IDE. :)
Happy that you got the idea and confirmed that both cases are failing.

[Tiller]

I tried something new:

    public Mono<ServerResponse> operationPost(final ServerRequest request) {
        return request
                .exchange()
                .getRequest()
                .getBody()
                .doOnNext(System.err::println)
                .map(b -> b.toString(StandardCharsets.UTF_8))
                .doOnNext(System.err::println)
                // .last()
                .next()
                .timeout(Duration.ofSeconds(5l))
                .doOnNext(System.err::println)
                .then(ServerResponse.ok().bodyValue(List.of()));
    }

And it works! And it prints:

UnpooledSlicedByteBuf(ridx: 0, widx: 17, cap: 17/17, unwrapped: PooledUnsafeDirectByteBuf(ridx: 121, widx: 121, cap: 1024))
{"hello":"world"}

However, switching .next() to .last() make the timeout reappear

But it still prints the same output:

UnpooledSlicedByteBuf(ridx: 0, widx: 17, cap: 17/17, unwrapped: PooledUnsafeDirectByteBuf(ridx: 121, widx: 121, cap: 1024))
{"hello":"world"}

So for some reasons, the Flux returned by getBody() never gets completed

[Tiller]

Investigating further:
With nghttp, FluxReceive subscription is completed in reactor.netty.channel.FluxReceive.terminateReceiver
With nodejs, it is never completed

[violetagg]

@Tiller hold for now I think I found the issue but need some time to reproduce it with Reactor Netty only

[Tiller]

Ok, I'll let you investigate further :)
Just last comment: with nghttp it first closes the inbound, then drainReceiver is called (which allows the completion of the flux), but with nodejs drainReceiver is called first, then the inbound is closed. But as there are no more calls to drainReceiver, it nevers closes the flux

[violetagg]

@Tiller if you are able to build Reactor Netty locally can you test #1052

[Tiller]

It works! Thanks :)

[Tiller]

Just fyi, I tried to send 3 DATA frames with a fragmented body, and I correctly received the whole body with all my chunks reunited