Vertx-web: BodyHandler with clustered vertx crashing with already ended request?

[apatrida]

I have clustered vertx on local network, one verticle does nothing at all, another is vertx-web.

When clustered, body handlers all crash with:

 Request has already been read
java.lang.IllegalStateException: Request has already been read
    at io.vertx.core.http.impl.HttpServerRequestImpl.checkEnded(HttpServerRequestImpl.java:409) ~[vertx-core-3.0.0.jar:na]
    at io.vertx.core.http.impl.HttpServerRequestImpl.setExpectMultipart(HttpServerRequestImpl.java:310) ~[vertx-core-3.0.0.jar:na]
    at io.vertx.ext.web.handler.impl.BodyHandlerImpl$BHandler.<init>(BodyHandlerImpl.java:86) ~[vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.handler.impl.BodyHandlerImpl$BHandler.<init>(BodyHandlerImpl.java:75) ~[vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.handler.impl.BodyHandlerImpl.handle(BodyHandlerImpl.java:49) ~[vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.handler.impl.BodyHandlerImpl.handle(BodyHandlerImpl.java:36) ~[vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RouteImpl.handleContext(RouteImpl.java:218) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:67) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:96) [vertx-web-3.0.0.jar:na]
    at com.collokia.web.api.ApiVerticle$start$3$1.handle(ApiVerticle.kt:84) [main/:na]
    at com.collokia.web.api.ApiVerticle$start$3$1.handle(ApiVerticle.kt:54) [main/:na]
    at io.vertx.ext.web.impl.RouteImpl.handleContext(RouteImpl.java:218) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:67) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:96) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.handler.impl.CorsHandlerImpl.handle(CorsHandlerImpl.java:142) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.handler.impl.CorsHandlerImpl.handle(CorsHandlerImpl.java:38) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RouteImpl.handleContext(RouteImpl.java:218) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:67) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:96) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.handler.impl.UserSessionHandlerImpl.handle(UserSessionHandlerImpl.java:63) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.handler.impl.UserSessionHandlerImpl.handle(UserSessionHandlerImpl.java:29) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RouteImpl.handleContext(RouteImpl.java:218) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:67) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:96) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.handler.impl.SessionHandlerImpl.lambda$handle$130(SessionHandlerImpl.java:96) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.handler.impl.SessionHandlerImpl$$Lambda$66/1550262877.handle(Unknown Source) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.sstore.impl.ClusteredSessionStoreImpl.lambda$null$140(ClusteredSessionStoreImpl.java:54) [vertx-web-3.0.0.jar:na]
    at io.vertx.ext.web.sstore.impl.ClusteredSessionStoreImpl$$Lambda$74/565770072.handle(Unknown Source) [vertx-web-3.0.0.jar:na]
    at io.vertx.core.impl.FutureImpl.checkCallHandler(FutureImpl.java:135) [vertx-core-3.0.0.jar:na]
    at io.vertx.core.impl.FutureImpl.setHandler(FutureImpl.java:100) [vertx-core-3.0.0.jar:na]
    at io.vertx.core.impl.ContextImpl.lambda$null$13(ContextImpl.java:288) [vertx-core-3.0.0.jar:na]
    at io.vertx.core.impl.ContextImpl$$Lambda$37/617001915.handle(Unknown Source) [vertx-core-3.0.0.jar:na]
    at io.vertx.core.impl.ContextImpl.lambda$wrapTask$15(ContextImpl.java:314) [vertx-core-3.0.0.jar:na]
    at io.vertx.core.impl.ContextImpl$$Lambda$38/1488040209.run(Unknown Source) [vertx-core-3.0.0.jar:na]
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:357) [netty-common-4.0.28.Final.jar:4.0.28.Final]
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:357) [netty-transport-4.0.28.Final.jar:4.0.28.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111) [netty-common-4.0.28.Final.jar:4.0.28.Final]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_40]

When not clustered, works fine.

[apatrida]

clustered = with hazelcast cluster manager

[apatrida]

Does clustering cause some implied body handler?

If so I can try a MaybeBodyHandler implementation like:

val bodyHandler =  BodyHandler.create().setBodyLimit(8 * 1024)
            router.route().method(HttpMethod.PUT).method(HttpMethod.POST).method(HttpMethod.PATCH).handler { routingContext ->
  if (!routingContext.request().isEnded()) {
      bodyHandler.handle(routingContext)
   } else {
      routingContext.next()
   }
  }

[apatrida]

Nope, with that the body is just ended but empty. why is it ended?

[apatrida]

Even cluster without any other nodes fails.

[apatrida]

the framework adding body handler later was after the clustering code. I think the clustering code or something there does async call or something that means the body handler isn't adding the internal body capture class until after the body started to come, so it missed grabbing it.

putting the BodyHandler as the first handler helps. Then again if you are doing something before you want to capture a body handler, you have to suspend/resume?

[purplefox]

Can you provide a reproducer please?

[pmlopes]

@apatrida you should have the body handler as close as possible to the beginning of your main router, the reason is that it needs to get called before any async code is called. The reason for this is because when the request starts to be processed by web the body handler must instruct netty to enable multipart parsing and setup the handler that will then consume the payload into web's context.

If you do this after the executing leaves the main loop then netty assumes that you're not interested in the body and when you instruct it later you will get the exception that the body has been read, which in this case was ignored.

We should document this better indeed.

[meggarr]

Hi there,

I met a similar issue today

2015-10-13 17:20:52.307 ERROR [vert.x-eventloop-thread-6] RoutingContextImplBase - Unexpected exception in route
java.lang.IllegalStateException: Request has already been read
        at io.vertx.core.http.impl.HttpServerRequestImpl.checkEnded(HttpServerRequestImpl.java:414) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.core.http.impl.HttpServerRequestImpl.setExpectMultipart(HttpServerRequestImpl.java:315) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.impl.HttpServerRequestWrapper.setExpectMultipart(HttpServerRequestWrapper.java:146) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.handler.impl.BodyHandlerImpl$BHandler.<init>(BodyHandlerImpl.java:86) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.handler.impl.BodyHandlerImpl$BHandler.<init>(BodyHandlerImpl.java:75) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.handler.impl.BodyHandlerImpl.handle(BodyHandlerImpl.java:49) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.handler.impl.BodyHandlerImpl.handle(BodyHandlerImpl.java:36) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.impl.RouteImpl.handleContext(RouteImpl.java:220) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:78) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:93) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.handler.impl.UserSessionHandlerImpl.handle(UserSessionHandlerImpl.java:63) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.handler.impl.UserSessionHandlerImpl.handle(UserSessionHandlerImpl.java:29) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.impl.RouteImpl.handleContext(RouteImpl.java:220) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.impl.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:78) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.impl.RoutingContextImpl.next(RoutingContextImpl.java:93) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.handler.impl.SessionHandlerImpl.lambda$handle$45(SessionHandlerImpl.java:117) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.handler.impl.SessionHandlerImpl.lambda$doGetSession$47(SessionHandlerImpl.java:142) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.sstore.impl.LocalSessionStoreImpl.get(LocalSessionStoreImpl.java:55) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.handler.impl.SessionHandlerImpl.doGetSession(SessionHandlerImpl.java:130) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.ext.web.handler.impl.SessionHandlerImpl.lambda$null$46(SessionHandlerImpl.java:137) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.core.impl.VertxImpl$InternalTimerHandler.handle(VertxImpl.java:749) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.core.impl.VertxImpl$InternalTimerHandler.handle(VertxImpl.java:720) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.vertx.core.impl.ContextImpl.lambda$wrapTask$16(ContextImpl.java:335) ~[sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:358) [sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:357) [sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112) [sam-provider-DEV-SNAPSHOT-fat.jar:?]
        at java.lang.Thread.run(Thread.java:745) [?:1.8.0_60]

My code is something like,

    Router router = Router.router(vertx);

    // - Session w/ local store
    router.route().handler(CookieHandler.create());
    SessionStore store = LocalSessionStore.create(vertx);
    router.route().handler(SessionHandler.create(store)
        .setSessionCookieName(SESSION_COOKIE_NAME).setSessionTimeout(sessionTimeout * 1000));
    router.route().handler(UserSessionHandler.create(authProvider));

    // - Body & x-response-time
    router.route().handler(BodyHandler.create());
    router.route().handler(ResponseTimeHandler.create());

    router.post("/login").handler(this::handleLogin);

Would it move the bodyHandler ahead before the cookie handler?

And, BTW, when is the Netty body multipart parser starting?

Thanks in advance!

[pmlopes]

As replied before you should move the body handler closest to the top and the Netty body multipart parser starts when the body handlers is executed, this is why it should be ASAP once you enter a callback then you're too late.

Since this was a question rather than an issue, it would be better to ask on the forum instead than here, so we can share the knowledge and the community can help you even faster than just the committers.

[meggarr]

Thanks, and I will put my questions in Google Group next time.

[mhstnsc]

@meggarr Also i wonder why should i put the body handler before the SessionHandler?
Unfortunatelly the HttpServer keeps pouring data into the application so if the BodyHandler is not registered then this data is either incomplete or lost (if the request is small enough, then it throws exception.)

@pmlopes @purplefox I hit this problem today and i think this bug should be reopened. IMO it is a critical aspect of the whole thing.

It sounds natural to me to first authorize a request and then start reading the body in memory or otherwise put the application should have full control when to accept processing new data. TCP protocol and kernel supports backpressure so why hide this feature.

What if the body is processed in a streamed way without fully loading it in memory, how is one supposed to handle this without having to implement some complicated mechanisms (dump it on disk or something) while the backpressure mechanism exists.

Is there a way to put a backpressure on the HttpServer? Http server should not push data but let the vertx-web pull data when is ready to do so or otherwise put HttpServerRequest.handler method should have an asynchronous signature. I am not familliar with netty but i will try to find out if this is possible.

An attacker could exploit this to perform a DOS since it could launch many invalid requests and since the app has to digest all the data or loose it....

[mhstnsc]

Found it. API is already there but i did not play with it yet. The service handler can register itself before the body handler but in this case it must do a HttpServerRequest.pause or just HttpServerRequest.netsocket().pause() to stop flux of data

[pmlopes]

@mstanescu the reason why body handler needs to be high in the order has been already explained but I'll do it again. As you know vert.x uses netty for all network I/O related tasks and the way netty works regarding http body parsing is in a ultra simplistic way like this:

• a request from a client is initiated
• netty will call vert.x request handler
• before the previous call returns (and here returns means that the vert.x handler reaches the return statement) the netty handler needs to get to know if it should attempt to parse the body payload or just skip it.
• If you have some handler that does some async call, say auth using DB then all remaining handlers will be delayed and only called after the async call completes, which means that the return as already been called so the handler did not request netty to parse the body.
• If you need to do validations before allowing parsing the body you have 2 choices:
  
  
  
  • either do basic in memory validations, such as JWT tokens, which can be validated without need to call async methods
  
  
  • pause/resume the request parsing
  
  

Regarding the other question, there is no pump between core and web so back pressure between the 2 does not apply.

If you want to queue all requests to avoid DoS attacks then you might just moving your failure somewhere else since the same attacker will now not strangle your processing handler but exhaust your RAM by filling up your queue.

To mitigate DoS attacks you will need proper network monitoring, probably filter request/clients at a firewall level, listen to the thread blocked alerts which can be both a syntom of blocking code or too much load on the CPU that a handler is not getting enough CPU time to complete, etc... etc... it is a subject on it own!!!