Buffered JSON parsing (#8803)

Currently, we use a non-blocking parser to transform input JSON to a JsonNode, then send it downstream, and finally convert it to the desired request type when the route arguments are bound. This patch delays parsing until the conversion to the route type. It consists of these parts:

- A change to JsonContentProcessor. The JsonContentProcessor has to handle certain scenarios (`application/x-json-stream`, or our support for streaming a json array as a Publisher) where the input contains multiple json tokens, Instead of forwarding the input bytes to an async parser, they are first split into individual json values, that are then forwarded or parsed in bulk.
- A new JsonCounter class. This class handles splitting the input into its individual json nodes. It is basically a limited JSON parser that counts braces. It is already very optimized, but it also has a special optimization when the content type is `application/json`, in which case it assumes only a single json value in the input (this is technically a breaking change, but breaks no tests). I also wrote fuzz tests for JsonCounter, I will submit them in a separate PR to https://github.com/micronaut-projects/micronaut-fuzzing .
- A benchmark. It's not run by the build tools, but it is good to have around.
- An option in NettyHttpServerConfiguration to re-enable eager parsing to JsonNode. JsonCounter is still used, so it's not 100% the old behavior, but it should be more compatible in some respects because it doesn't change the conversion logic anymore.
- A new LazyJsonNode class. It contains a ByteBuffer with the actual unparsed bytes. It's a bit complicated because it contains a reference counted buffer that has to be released after conversion, but also sometimes multiple converters are called for the same LazyJsonNode (once to JsonNode, once to a user-defined type). To solve this, when there is a conversion to JsonNode, it keeps that JsonNode around (and releases the buffer). If there are future conversions to a specific type, it will use the JsonNode as the source instead.
- A new JsonSyntaxException class. JsonMappers can throw this to signal a syntax error in the JSON, which will be reported differently than a data binding error.
- Changes to JsonMapper API: The asynchronous parser is deprecated. There is a new readValue method that takes a ByteBuffer. The jackson implementation has an optimization for netty ByteBuffers.
- New converters for LazyJsonNode. Also removed some old superfluous converters for JsonNode.
- A change to RequestLifecycle to show JSON syntax error messages like we did previously, instead of the opaque 400 that we usually send for conversion errors.
- Some test changes to reflect changes in error messages. Because we now parse the input in bulk, in some cases jackson can decorate errors with short snippets of the failing input data. Is this an issue?

There are a few potential incompatibilities with this change:

- Non-standard JSON features when combined with JsonCounter (i.e. when streaming a JSON array, or when using `application/x-json-stream`) will break even when the JSON parser is configured to support them. e.g. if the user configured jackson to ignore comments, that may fail now.
- When the input body is never bound, JSON syntax errors may not be caught.
- JsonCounter only supports UTF-8, by design. This is permitted by the JSON standard, however Jackson also supports UTF-16 and UTF-32. imo this is a net benefit, to avoid potential parser differential vulnerabilities. I also made a jackson feature to match this, though JsonCounter is sufficient now: FasterXML/jackson-core#921
- JsonMapper implementations that do not use JsonSyntaxException yet will have less verbose HTTP errors until they switch to JsonSyntaxException.

Author: yawkat

buffer-netty/src/main/java/io/micronaut/buffer/netty/NettyByteBuffer.java

@@ -50,7 +50,7 @@ class NettyByteBuffer implements ByteBuffer<ByteBuf>, ReferenceCounted {
     }
 
     @Override
-    public ByteBuffer retain() {
+    public NettyByteBuffer retain() {
         delegate.retain();
         return this;
     }

core/src/main/java/io/micronaut/core/io/buffer/ReferenceCounted.java

@@ -25,7 +25,7 @@ public interface ReferenceCounted {
      *
      * @return this
      */
-    ByteBuffer retain();
+    ReferenceCounted retain();
 
     /**
      * Release a reference to this object.

http-server-netty/build.gradle

@@ -31,6 +31,10 @@ dependencies {
     compileOnly libs.kotlin.stdlib
     compileOnly libs.managed.netty.transport.native.unix.common
 
+    testImplementation 'org.openjdk.jmh:jmh-core:1.36'
+    testAnnotationProcessor 'org.openjdk.jmh:jmh-generator-annprocess:1.36'
+
+
     testCompileOnly project(":inject-groovy")
     testCompileOnly(libs.jetbrains.annotations)
     testAnnotationProcessor project(":inject-java")

http-server-netty/src/main/java/io/micronaut/http/server/netty/configuration/NettyHttpServerConfiguration.java

@@ -114,6 +114,22 @@ public class NettyHttpServerConfiguration extends HttpServerConfiguration {
     @SuppressWarnings("WeakerAccess")
     public static final boolean DEFAULT_KEEP_ALIVE_ON_SERVER_ERROR = true;
 
+    /**
+     * The default value for eager parsing.
+     *
+     * @since 4.0.0
+     */
+    @SuppressWarnings("WeakerAccess")
+    public static final boolean DEFAULT_EAGER_PARSING = false;
+
+    /**
+     * The default value for eager parsing.
+     *
+     * @since 4.0.0
+     */
+    @SuppressWarnings("WeakerAccess")
+    public static final int DEFAULT_JSON_BUFFER_MAX_COMPONENTS = 4096;
+
     private static final Logger LOG = LoggerFactory.getLogger(NettyHttpServerConfiguration.class);
 
     private final List<ChannelPipelineListener> pipelineCustomizers;
@@ -140,6 +156,8 @@ public class NettyHttpServerConfiguration extends HttpServerConfiguration {
     private boolean keepAliveOnServerError = DEFAULT_KEEP_ALIVE_ON_SERVER_ERROR;
     private String pcapLoggingPathPattern = null;
     private List<NettyListenerConfiguration> listeners = null;
+    private boolean eagerParsing = DEFAULT_EAGER_PARSING;
+    private int jsonBufferMaxComponents = DEFAULT_JSON_BUFFER_MAX_COMPONENTS;
 
     /**
      * Default empty constructor.
@@ -564,6 +582,51 @@ public void setListeners(List<NettyListenerConfiguration> listeners) {
         this.listeners = listeners;
     }
 
+    /**
+     * Parse incoming JSON data eagerly, before route binding. Default value
+     * {@value DEFAULT_EAGER_PARSING}.
+     *
+     * @return Whether to parse incoming JSON data eagerly before route binding
+     * @since 4.0.0
+     */
+    public boolean isEagerParsing() {
+        return eagerParsing;
+    }
+
+    /**
+     * Parse incoming JSON data eagerly, before route binding. Default value
+     * {@value DEFAULT_EAGER_PARSING}.
+     *
+     * @param eagerParsing Whether to parse incoming JSON data eagerly before route binding
+     * @since 4.0.0
+     */
+    public void setEagerParsing(boolean eagerParsing) {
+        this.eagerParsing = eagerParsing;
+    }
+
+    /**
+     * Maximum number of buffers to keep around in JSON parsing before they should be consolidated.
+     * Defaults to {@value #DEFAULT_JSON_BUFFER_MAX_COMPONENTS}.
+     *
+     * @return The maximum number of components
+     * @since 4.0.0
+     */
+    public int getJsonBufferMaxComponents() {
+        return jsonBufferMaxComponents;
+    }
+
+
+    /**
+     * Maximum number of buffers to keep around in JSON parsing before they should be consolidated.
+     * Defaults to {@value #DEFAULT_JSON_BUFFER_MAX_COMPONENTS}.
+     *
+     * @param jsonBufferMaxComponents The maximum number of components
+     * @since 4.0.0
+     */
+    public void setJsonBufferMaxComponents(int jsonBufferMaxComponents) {
+        this.jsonBufferMaxComponents = jsonBufferMaxComponents;
+    }
+
     /**
      * Http2 settings.
      */

http-server-netty/src/main/java/io/micronaut/http/server/netty/jackson/JsonContentProcessor.java

@@ -15,24 +15,24 @@
  */
 package io.micronaut.http.server.netty.jackson;
 
+import io.micronaut.buffer.netty.NettyByteBufferFactory;
 import io.micronaut.core.annotation.Internal;
 import io.micronaut.core.async.publisher.Publishers;
-import io.micronaut.core.async.subscriber.CompletionAwareSubscriber;
+import io.micronaut.core.io.buffer.ByteBuffer;
 import io.micronaut.core.type.Argument;
 import io.micronaut.http.MediaType;
-import io.micronaut.http.server.HttpServerConfiguration;
 import io.micronaut.http.server.netty.AbstractHttpContentProcessor;
 import io.micronaut.http.server.netty.HttpContentProcessor;
 import io.micronaut.http.server.netty.NettyHttpRequest;
+import io.micronaut.http.server.netty.configuration.NettyHttpServerConfiguration;
 import io.micronaut.json.JsonMapper;
+import io.micronaut.json.convert.LazyJsonNode;
 import io.micronaut.json.tree.JsonNode;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufHolder;
-import io.netty.buffer.ByteBufUtil;
-import io.netty.util.ReferenceCountUtil;
-import org.reactivestreams.Processor;
-import org.reactivestreams.Subscription;
+import io.netty.buffer.CompositeByteBuf;
 
+import java.io.IOException;
 import java.util.Collection;
 import java.util.Optional;
 
@@ -47,9 +47,8 @@
 public class JsonContentProcessor extends AbstractHttpContentProcessor {
 
     private final JsonMapper jsonMapper;
-    private Processor<byte[], JsonNode> jacksonProcessor;
-    private Collection<Object> out;
-    private Throwable failure = null;
+    private final JsonCounter counter = new JsonCounter();
+    private CompositeByteBuf buffer;
 
     /**
      * @param nettyHttpRequest The Netty Http request
@@ -58,111 +57,106 @@ public class JsonContentProcessor extends AbstractHttpContentProcessor {
      */
     public JsonContentProcessor(
             NettyHttpRequest<?> nettyHttpRequest,
-            HttpServerConfiguration configuration,
+            NettyHttpServerConfiguration configuration,
             JsonMapper jsonMapper) {
         super(nettyHttpRequest, configuration);
         this.jsonMapper = jsonMapper;
+
+        if (hasContentType(MediaType.APPLICATION_JSON_TYPE)) {
+
+            // if the content type is application/json, we can only have one root-level value
+            counter.noTokenization();
+        }
     }
 
     @Override
     public HttpContentProcessor resultType(Argument<?> type) {
-        boolean streamArray = false;
 
-        boolean isJsonStream = nettyHttpRequest.getContentType()
-            .map(mediaType -> mediaType.equals(MediaType.APPLICATION_JSON_STREAM_TYPE))
-            .orElse(false);
+        boolean isJsonStream = hasContentType(MediaType.APPLICATION_JSON_STREAM_TYPE);
 
         if (type != null) {
             Class<?> targetType = type.getType();
             if (Publishers.isConvertibleToPublisher(targetType) && !Publishers.isSingle(targetType)) {
                 Optional<Argument<?>> genericArgument = type.getFirstTypeVariable();
                 if (genericArgument.isPresent() && !Iterable.class.isAssignableFrom(genericArgument.get().getType()) && !isJsonStream) {
                     // if the generic argument is not a iterable type them stream the array into the publisher
-                    streamArray = true;
+                    counter.unwrapTopLevelArray();
                 }
             }
         }
+        return this;
+    }
 
-        this.jacksonProcessor = jsonMapper.createReactiveParser(p -> {
-        }, streamArray);
-        this.jacksonProcessor.subscribe(new CompletionAwareSubscriber<>() {
-
-            @Override
-            protected void doOnSubscribe(Subscription jsonSubscription) {
-                jsonSubscription.request(Long.MAX_VALUE);
-            }
-
-            @Override
-            protected void doOnNext(JsonNode message) {
-                if (out == null) {
-                    throw new IllegalStateException("Concurrent access not allowed");
-                }
-                out.add(message);
-            }
-
-            @Override
-            protected void doOnError(Throwable t) {
-                if (out == null) {
-                    throw new IllegalStateException("Concurrent access not allowed");
-                }
-                failure = t;
-            }
+    private boolean hasContentType(MediaType expected) {
+        Optional<MediaType> actual = nettyHttpRequest.getContentType();
+        return actual.isPresent() && actual.get().equals(expected);
+    }
 
-            @Override
-            protected void doOnComplete() {
-                if (out == null) {
-                    throw new IllegalStateException("Concurrent access not allowed");
-                }
-            }
-        });
-        this.jacksonProcessor.onSubscribe(new Subscription() {
-            @Override
-            public void request(long n) {
+    @Override
+    protected void onData(ByteBufHolder message, Collection<Object> out) throws Throwable {
+        ByteBuf content = message.content();
+        try {
+            countLoop(out, content);
+        } catch (Exception e) {
+            if (this.buffer != null) {
+                this.buffer.release();
+                this.buffer = null;
             }
+            throw e;
+        } finally {
+            content.release();
+        }
+    }
 
-            @Override
-            public void cancel() {
-                // happens on error, ignore
+    private void countLoop(Collection<Object> out, ByteBuf content) throws IOException {
+        long initialPosition = counter.position();
+        long bias = initialPosition - content.readerIndex();
+        while (content.isReadable()) {
+            counter.feed(content);
+            JsonCounter.BufferRegion bufferRegion = counter.pollFlushedRegion();
+            if (bufferRegion != null) {
+                long start = Math.max(initialPosition, bufferRegion.start());
+                flush(out, content.retainedSlice(
+                    Math.toIntExact(start - bias),
+                    Math.toIntExact(bufferRegion.end() - start)
+                ));
             }
-        });
-        return this;
+        }
+        if (counter.isBuffering()) {
+            int currentBufferStart = Math.toIntExact(Math.max(initialPosition, counter.bufferStart()) - bias);
+            bufferForNextRun(content.retainedSlice(currentBufferStart, content.writerIndex() - currentBufferStart));
+        }
     }
 
-    @Override
-    protected void onData(ByteBufHolder message, Collection<Object> out) throws Throwable {
-        if (jacksonProcessor == null) {
-            resultType(null);
+    private void bufferForNextRun(ByteBuf buffer) {
+        if (this.buffer == null) {
+            // number of components should not be too small to avoid unnecessary consolidation
+            this.buffer = buffer.alloc().compositeBuffer(((NettyHttpServerConfiguration) configuration).getJsonBufferMaxComponents());
         }
+        this.buffer.addComponent(true, buffer);
+    }
 
-        this.out = out;
-        ByteBuf content = message.content();
-        try {
-            byte[] bytes = ByteBufUtil.getBytes(content);
-            jacksonProcessor.onNext(bytes);
-        } finally {
-            ReferenceCountUtil.release(content);
-            this.out = null;
+    private void flush(Collection<Object> out, ByteBuf completedNode) throws IOException {
+        if (this.buffer != null) {
+            completedNode = completedNode == null ? this.buffer : this.buffer.addComponent(true, completedNode);
+            this.buffer = null;
         }
-        Throwable f = failure;
-        if (f != null) {
-            failure = null;
-            throw f;
+        ByteBuffer<ByteBuf> wrapped = NettyByteBufferFactory.DEFAULT.wrap(completedNode);
+        if (((NettyHttpServerConfiguration) configuration).isEagerParsing()) {
+            try {
+                out.add(jsonMapper.readValue(wrapped, Argument.of(JsonNode.class)));
+            } finally {
+                completedNode.release();
+            }
+        } else {
+            out.add(new LazyJsonNode(wrapped));
         }
     }
 
     @Override
     public void complete(Collection<Object> out) throws Throwable {
-        if (jacksonProcessor == null) {
-            resultType(null);
-        }
-
-        this.out = out;
-        jacksonProcessor.onComplete();
-        this.out = null;
-        Throwable f = failure;
-        if (f != null) {
-            failure = null;
-            throw f;
+        if (this.buffer != null) {
+            flush(out, null);
         }
     }
 }