Dynamic Request Body Size Limiting Using ContainerRequestFilter Not Closing Stream Immediately

[cdxf]

I'm attempting to implement a filter that dynamically limits the request body size, as an alternative to using the quarkus.http.limits.max-body-size configuration property:

@Provider
@Priority(Priorities.ENTITY_CODER + 100) // Run after Content-Length check
public class InputStreamLimitFilter implements ContainerRequestFilter {

    public static class LimitedInputStream extends InputStream {

        private final InputStream delegate;
        private final long limit;
        private final ContainerRequestContext requestContext;
        private long bytesRead;

        public LimitedInputStream(InputStream delegate, ContainerRequestContext requestContext, long limit) {
            this.requestContext = requestContext;
            this.delegate = delegate;
            this.limit = limit;
            this.bytesRead = 0;
        }

        @Override
        public int read() throws IOException {
            if (bytesRead >= limit) {
                delegate.close();
                throw new RuntimeException("Limit exceeded");
            }
            int b = delegate.read();
            if (b != -1) {
                bytesRead++;
            }
            return b;
        }

        @Override
        public int read(byte[] b, int off, int len) throws IOException {
            if (bytesRead >= limit) {
                System.out.println("Limit exceeded");
                delegate.close();
                throw new RuntimeException("Limit exceeded");
            }
            // Important: Limit the number of bytes to read
            long remaining = limit - bytesRead;
            if (remaining < len) {
                len = (int) remaining;
            }

            int bytesReadNow = delegate.read(b, off, len);
            if (bytesReadNow != -1) {
                bytesRead += bytesReadNow;
            }
            return bytesReadNow;
        }

        @Override
        public long skip(long n) throws IOException {

            // Important: Limit the number of bytes to skip
            long remaining = limit - bytesRead;
            if (remaining < n) {
                n = remaining;
            }

            long bytesSkipped = delegate.skip(n);
            bytesRead += bytesSkipped;
            return bytesSkipped;
        }

        @Override
        public int available() throws IOException {

            long remaining = limit - bytesRead;
            int available = delegate.available();

            // Return the smaller of the two
            return (int) Math.min(remaining, available);
        }

        @Override
        public int read(byte[] b) throws IOException {
            if (bytesRead >= limit) {
                delegate.close();
                System.out.println("Limit exceeded");
                throw new RuntimeException("Limit exceeded");
            }

            int bytesReadNow = delegate.read(b);
            if (bytesReadNow != -1) {
                bytesRead += bytesReadNow;
            }
            return bytesReadNow;
        }

        @Override
        public byte[] readAllBytes() throws IOException {
            if (bytesRead >= limit) {
                delegate.close();
                System.out.println("Limit exceeded");
                throw new RuntimeException("Limit exceeded");
            }

            long remaining = limit - bytesRead;
            if (remaining <= 0) {
                return new byte[0];
            }

            byte[] buffer = delegate.readNBytes((int) remaining);
            bytesRead += buffer.length;
            return buffer;
        }

        @Override
        public byte[] readNBytes(int len) throws IOException {
            if (bytesRead >= limit) {
                delegate.close();
                System.out.println("Limit exceeded");
                throw new RuntimeException("Limit exceeded");
            }

            long remaining = limit - bytesRead;
            if (remaining < len) {
                len = (int) remaining;
            }

            byte[] buffer = delegate.readNBytes(len);
            bytesRead += buffer.length;
            return buffer;
        }

        @Override
        public int readNBytes(byte[] b, int off, int len) throws IOException {

            if (bytesRead >= limit) {
                delegate.close();
                throw new RuntimeException("Limit exceeded");
            }

            long remaining = limit - bytesRead;
            if (remaining < len) {
                len = (int) remaining;
            }

            int bytesReadNow = delegate.readNBytes(b, off, len);
            bytesRead += bytesReadNow;
            return bytesReadNow;
        }

        @Override
        public void skipNBytes(long n) throws IOException {
            if (bytesRead >= limit) {
                delegate.close();
                throw new RuntimeException("Limit exceeded");
            }

            long remaining = limit - bytesRead;
            if (remaining < n) {
                respondWithLimitExceeded();
            }

            delegate.skipNBytes(n);
            bytesRead += n;
        }

        @Override
        public void close() throws IOException {
            delegate.close();
        }

        @Override
        public void mark(int readlimit) {
            delegate.mark(readlimit);
        }

        @Override
        public void reset() throws IOException {
            delegate.reset();
        }

        @Override
        public boolean markSupported() {
            return delegate.markSupported();
        }

        @Override
        public long transferTo(OutputStream out) throws IOException {
            if (bytesRead >= limit) {
                delegate.close();
                throw new RuntimeException("Limit exceeded");
            }

            long remaining = limit - bytesRead;
            long transferred = delegate.transferTo(out);
            if (transferred > remaining) {
                respondWithLimitExceeded();
            }

            bytesRead += transferred;
            return transferred;
        }
        private void respondWithLimitExceeded() {
            requestContext.abortWith(Response.status(Response.Status.BAD_REQUEST)
                    .entity("Request body exceeds the allowed size of " + limit + " bytes.")
                    .build());
        }
    }

    @Override
    public void filter(ContainerRequestContext requestContext) throws IOException {
        long maxAllowedSize = 1024 * 1024L;
        LimitedInputStream limitedStream = new LimitedInputStream(requestContext.getEntityStream(),  requestContext, maxAllowedSize);
        requestContext.setEntityStream(limitedStream);
    }
}

When I upload a big file, the server response like I expected:

{
"details": "Error id 97f8fb04-b73f-419b-b551-272faa25dcd7-22, java.io.IOException: Stream is closed",
"stack": "java.io.IOException: Stream is closed\r\n\tat org.jboss.resteasy.reactive.server.vertx.VertxInputStream.read(VertxInputStream.java:79)\r\n\tat org.jboss.resteasy.reactive.server.vertx.VertxInputStream.read(VertxInputStream.java:73)\r\n\tat org.jboss.resteasy.reactive.server.core.multipart.MultiPartParserDefinition$MultiPartUploadHandler.parseBlocking(MultiPartParserDefinition.java:246)\r\n\tat org.jboss.resteasy.reactive.server.handlers.FormBodyHandler.handle(FormBodyHandler.java:95)\r\n\tat io.quarkus.resteasy.reactive.server.runtime.QuarkusResteasyReactiveRequestContext.invokeHandler(QuarkusResteasyReactiveRequestContext.java:150)\r\n\tat org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.run(AbstractResteasyReactiveContext.java:147)\r\n\tat io.quarkus.vertx.core.runtime.VertxCoreRecorder$15.runWith(VertxCoreRecorder.java:637)\r\n\tat org.jboss.threads.EnhancedQueueExecutor$Task.doRunWith(EnhancedQueueExecutor.java:2675)\r\n\tat org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2654)\r\n\tat org.jboss.threads.EnhancedQueueExecutor.runThreadBody(EnhancedQueueExecutor.java:1627)\r\n\tat org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1594)\r\n\tat org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:11)\r\n\tat org.jboss.threads.ThreadLocalResettingRunnable.run(ThreadLocalResettingRunnable.java:11)\r\n\tat io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)\r\n\tat java.base/java.lang.Thread.run(Thread.java:1623)"
}

Problem:

The server only returns a response after the entire file has been uploaded. This behavior differs from using quarkus.http.limits.max-body-size, which immediately closes the stream upon exceeding the limit.

Suspected Cause:

The FormBodyHandler implementation uses CapturingInputStream, which reads the entire stream before processing. Consequently, attempting to close the stream prematurely has no effect.

Can you tell me the cause of this problem, is there any solution for this?