Copilot commented on code in PR #3402:
URL: https://github.com/apache/tika/pull/3402#discussion_r4355627130
##########
tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/MaxRequestSizeFilter.java:
##########
@@ -19,47 +19,84 @@
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import jakarta.ws.rs.container.ContainerRequestContext;
import jakarta.ws.rs.container.ContainerRequestFilter;
+import jakarta.ws.rs.core.HttpHeaders;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.ext.ExceptionMapper;
import jakarta.ws.rs.ext.Provider;
+import org.apache.cxf.jaxrs.utils.JAXRSUtils;
+import org.apache.cxf.message.Message;
/**
- * Rejects request bodies larger than {@code maxRequestSizeBytes}.
+ * Rejects request bodies larger than {@code maxRequestSizeBytes} (413), still
arriving
+ * after {@code maxRequestReadMillis} (408), or that would take the bytes all
requests hold
+ * past {@code maxInFlightRequestBytes} (429, see {@link RequestBodyBudget}).
* <p>
* A declared Content-Length over the limit is refused before the body is
read. Requests
* without a usable Content-Length -- chunked transfer encoding, in particular
-- are
* counted as they are consumed, so the limit holds whether or not the client
is honest
- * about the size.
+ * about the size. The wire under the entity is bounded by
+ * {@link RequestReadDeadlineInterceptor}, whose start time this deadline
shares.
*/
@Provider
public class MaxRequestSizeFilter implements ContainerRequestFilter {
static final String TOO_LARGE_MESSAGE = "Request body exceeds
maxRequestSizeBytes";
+ static final String TOO_SLOW_MESSAGE = "Request body not received within
maxRequestReadMillis";
+ static final String IDLE_MESSAGE = "Request body stalled past
idleTimeoutMillis";
+ static final String BUSY_MESSAGE = "Request bodies in flight exceed
maxInFlightRequestBytes";
Review Comment:
A few of these messages are ambiguous or slightly misleading: (1)
BUSY_MESSAGE triggers when the request *would exceed* the budget, not when it
already does; consider wording like \"would exceed\" to better match behavior
and docs. (2) TOO_SLOW_MESSAGE (and similar messages elsewhere) later appends a
raw number without units; include `ms` in the formatted message to make log
lines/self-reported errors clearer.
##########
tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/MaxRequestSizeIntegrationTest.java:
##########
@@ -0,0 +1,606 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.server.standard;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.BufferedReader;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.zip.GZIPOutputStream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import jakarta.ws.rs.core.Response;
+import org.apache.cxf.jaxrs.client.WebClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.server.core.IntegrationTestBase;
+import org.apache.tika.server.core.TestPortAllocator;
+
+/**
+ * Transfer limits on a real server, across every body-accepting endpoint its
welcome page
+ * lists.
+ */
+public class MaxRequestSizeIntegrationTest extends IntegrationTestBase {
+
+ // above what /language reads, so that endpoint cannot trip the limit by
reading alone
+ private static final int MAX_BYTES = 200_000;
+ private static final int MAX_READ_MILLIS = 3_000;
+ private static final int MAX_WRITE_MILLIS = 3_000;
+ private static final int IDLE_TIMEOUT_MILLIS = 2_000;
+ // room for two maximum-size bodies and part of a third
+ private static final int MAX_IN_FLIGHT_BYTES = 3 * MAX_BYTES;
+ // past the 16 MiB CXF discards after a response, so a reply proves
reading stopped
+ private static final int SENT_BYTES = 20 * 1024 * 1024;
+ private static final String BOUNDARY = "tika-limit-boundary";
+ // from tika-server-core's test preset catalog
+ private static final String PRESET = "catalog-xml-content";
+ // text in bigTextZip(): far more than the socket buffers hold
+ private static final int TEXT_BYTES = 16 * 1024 * 1024;
+ private static final String[] TIKA_TEXT = {"PUT", "/tika/text",
"text/plain"};
+
+ private static final Map<String, byte[]> PREAMBLES = new LinkedHashMap<>();
+
+ static {
+ PREAMBLES.put("text/plain", new byte[0]);
+ // an open JSON string, so a JSON reader keeps consuming
+ PREAMBLES.put("application/json", "{\"a\":\"".getBytes(UTF_8));
+ PREAMBLES.put("multipart/form-data; boundary=" + BOUNDARY,
+ ("--" + BOUNDARY + "\r\nContent-Disposition: form-data;
name=\"file\"; "
+ + "filename=\"a.txt\"\r\nContent-Type:
text/plain\r\n\r\n").getBytes(UTF_8));
+ }
+
+ @TempDir
+ static Path classWorkingDir;
+
+ private final int metricsPort = TestPortAllocator.findFreePort();
+
+ @BeforeAll
+ public void startServer() throws Exception {
+ // every endpoint: the defaults, plus pipes/async via allowPipes and
the config routes
+ String config = """
+ {
Review Comment:
`@BeforeAll` methods must be `static` unless the test class uses
`@TestInstance(Lifecycle.PER_CLASS)`. As written, this will fail to run under
JUnit 5 default lifecycle. Make `startServer()` static or add
`@TestInstance(TestInstance.Lifecycle.PER_CLASS)` to the class.
##########
tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/MaxRequestSizeIntegrationTest.java:
##########
@@ -0,0 +1,606 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.server.standard;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.BufferedReader;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.zip.GZIPOutputStream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import jakarta.ws.rs.core.Response;
+import org.apache.cxf.jaxrs.client.WebClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.server.core.IntegrationTestBase;
+import org.apache.tika.server.core.TestPortAllocator;
+
+/**
+ * Transfer limits on a real server, across every body-accepting endpoint its
welcome page
+ * lists.
+ */
+public class MaxRequestSizeIntegrationTest extends IntegrationTestBase {
+
+ // above what /language reads, so that endpoint cannot trip the limit by
reading alone
+ private static final int MAX_BYTES = 200_000;
+ private static final int MAX_READ_MILLIS = 3_000;
+ private static final int MAX_WRITE_MILLIS = 3_000;
+ private static final int IDLE_TIMEOUT_MILLIS = 2_000;
+ // room for two maximum-size bodies and part of a third
+ private static final int MAX_IN_FLIGHT_BYTES = 3 * MAX_BYTES;
+ // past the 16 MiB CXF discards after a response, so a reply proves
reading stopped
+ private static final int SENT_BYTES = 20 * 1024 * 1024;
+ private static final String BOUNDARY = "tika-limit-boundary";
+ // from tika-server-core's test preset catalog
+ private static final String PRESET = "catalog-xml-content";
+ // text in bigTextZip(): far more than the socket buffers hold
+ private static final int TEXT_BYTES = 16 * 1024 * 1024;
+ private static final String[] TIKA_TEXT = {"PUT", "/tika/text",
"text/plain"};
+
+ private static final Map<String, byte[]> PREAMBLES = new LinkedHashMap<>();
+
+ static {
+ PREAMBLES.put("text/plain", new byte[0]);
+ // an open JSON string, so a JSON reader keeps consuming
+ PREAMBLES.put("application/json", "{\"a\":\"".getBytes(UTF_8));
+ PREAMBLES.put("multipart/form-data; boundary=" + BOUNDARY,
+ ("--" + BOUNDARY + "\r\nContent-Disposition: form-data;
name=\"file\"; "
+ + "filename=\"a.txt\"\r\nContent-Type:
text/plain\r\n\r\n").getBytes(UTF_8));
+ }
+
+ @TempDir
+ static Path classWorkingDir;
+
+ private final int metricsPort = TestPortAllocator.findFreePort();
+
+ @BeforeAll
+ public void startServer() throws Exception {
+ // every endpoint: the defaults, plus pipes/async via allowPipes and
the config routes
+ String config = """
+ {
+ "server": {
+ "allowPipes": true,
+ "allowPerRequestConfig": true,
+ "allowHttp2": true,
+ "maxRequestSizeBytes": %d,
+ "maxRequestReadMillis": %r,
+ "maxResponseWriteMillis": %w,
+ "idleTimeoutMillis": %i,
+ "maxInFlightRequestBytes": %c,
+ "metricsPort": %m
+ },
+ "pipes": { "numClients": 1, "forkedJvmArgs": ["-Xmx256m"] },
+ "plugin-roots": "%s"
+ }
+ """
+ .replace("%d", Integer.toString(MAX_BYTES))
+ .replace("%r", Integer.toString(MAX_READ_MILLIS))
+ .replace("%w", Integer.toString(MAX_WRITE_MILLIS))
+ .replace("%i", Integer.toString(IDLE_TIMEOUT_MILLIS))
+ .replace("%c", Integer.toString(MAX_IN_FLIGHT_BYTES))
+ .replace("%m", Integer.toString(metricsPort))
+ .replace("%s",
Paths.get("target/plugins").toAbsolutePath().toString().replace("\\", "/"));
+ Path configFile =
Files.writeString(classWorkingDir.resolve("request-limit.json"), config);
+ startClassProcess(new String[]{"-config", configFile.toString(),
"--preset", PRESET},
+ classWorkingDir);
+ }
+
+ /** Whatever a test did to the server, every request's share of the body
budget comes back. */
+ @AfterEach
+ public void assertBudgetDrained() throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ long inFlight;
+ do {
+ inFlight = bodyBytesInFlight();
+ if (inFlight != 0) {
+ Thread.sleep(50);
+ }
+ } while (inFlight != 0 && System.nanoTime() < deadline);
+ assertEquals(0, inFlight, "request body budget not released");
+ }
+
+ @Test
+ public void testEveryEndpointAnswersOverLimitChunkedBeforeBodyEnds()
throws Exception {
+ List<String[]> endpoints = bodyEndpoints();
+ assertTrue(endpoints.size() > 40, "welcome page listed too few
endpoints: " + endpoints.size());
Review Comment:
This hard-coded threshold makes the integration test brittle to legitimate
endpoint list changes (feature flags, module additions/removals,
servlet/container differences). Prefer asserting basic invariants (e.g.,
non-empty, or contains a small set of known endpoints), or making the threshold
configurable.
##########
tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/ResponseWriteDeadlineInterceptorTest.java:
##########
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.server.core;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import
org.apache.tika.server.core.ResponseWriteDeadlineInterceptor.DeadlineOutputStream;
+import
org.apache.tika.server.core.ResponseWriteDeadlineInterceptor.ResponseWriteTimeoutException;
+
+/**
+ * The deadline as a timer: a write blocked by a client that reads a trickle
(each byte resetting
+ * the idle timeout) must still end at the deadline, which a check between
writes cannot do.
+ */
+public class ResponseWriteDeadlineInterceptorTest {
+
+ private static final long DEADLINE_MILLIS = 200;
Review Comment:
The assertion `>= DEADLINE_MILLIS` with a 200ms deadline can be flaky on
loaded/slow CI due to timer granularity and scheduling jitter. Consider adding
tolerance (e.g., allow slightly earlier firing), or asserting within a range
(not too early / not unreasonably late) to keep the test stable.
##########
tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ResponseWriteDeadlineInterceptor.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.server.core;
+
+import java.io.FilterOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BooleanSupplier;
+
+import jakarta.servlet.http.HttpServletResponse;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.ext.ExceptionMapper;
+import jakarta.ws.rs.ext.Provider;
+import jakarta.ws.rs.ext.WriterInterceptor;
+import jakarta.ws.rs.ext.WriterInterceptorContext;
+import org.apache.cxf.jaxrs.utils.JAXRSUtils;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.transport.http.AbstractHTTPDestination;
+import org.eclipse.jetty.ee11.servlet.ServletContextRequest;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.util.thread.Scheduler;
+
+/**
+ * Aborts a response still being written {@code maxResponseWriteMillis} after
it is committed;
+ * the idle timeout alone resets on every trickle a slow reader accepts.
+ * <p>
+ * A timer, not a check between writes: one blocking write to a client that
accepts a byte at
+ * a time can otherwise outlast the deadline indefinitely. From commit, not
the first write:
+ * CXF drains the unread request body inside that write, which the read
deadline bounds.
+ * Aborted, not closed: closing would end a chunked body cleanly, passing a
truncated document
+ * off as complete.
+ */
+@Provider
+public class ResponseWriteDeadlineInterceptor implements WriterInterceptor {
+
+ private final long maxResponseWriteMillis;
+
+ /**
+ * @param maxResponseWriteMillis how long writing a response body may
take; negative disables it
+ */
+ public ResponseWriteDeadlineInterceptor(long maxResponseWriteMillis) {
+ this.maxResponseWriteMillis = maxResponseWriteMillis;
+ }
+
+ @Override
+ public void aroundWriteTo(WriterInterceptorContext context) throws
IOException {
+ Message message = JAXRSUtils.getCurrentMessage();
+ ServletContextRequest request = JettyRequests.of(message);
+ // without a Jetty connection there is nothing to abort; on HTTP/2 the
abort would end
+ // every stream on the connection (see allowHttp2)
+ if (maxResponseWriteMillis >= 0 && request != null
+ &&
request.getConnectionMetaData().getHttpVersion().getVersion() < 20) {
+ Message inMessage = message.getExchange().getInMessage();
+ BooleanSupplier committed =
inMessage.get(AbstractHTTPDestination.HTTP_RESPONSE)
+ instanceof HttpServletResponse response ?
response::isCommitted : () -> true;
+ DeadlineOutputStream out = new
DeadlineOutputStream(context.getOutputStream(),
+ maxResponseWriteMillis,
request.getComponents().getScheduler(), committed,
+ () ->
request.getConnectionMetaData().getConnection().getEndPoint()
+ .close(new
ResponseWriteTimeoutException("maxResponseWriteMillis exceeded")),
+ inMessage);
+ Request.addCompletionListener(request, failure -> out.complete());
+ context.setOutputStream(out);
+ // CXF closes the message's stream after the writer; ours skips
that once aborted
+ if (message.getContent(OutputStream.class) == out.getDelegate()) {
+ message.setContent(OutputStream.class, out);
+ }
+ try {
+ context.proceed();
+ } catch (ResponseWriteTimeoutException e) {
+ // the timer logged it and closed the connection; CXF would
log it again as an error
+ if (!out.isAborted()) {
+ throw e;
+ }
+ }
+ return;
+ }
+ context.proceed();
+ }
+
+ /**
+ * From the first write, waits for the response to commit, then arms the
deadline. The
+ * timer and the request's completion race to end it, and exactly one
wins: once completed
+ * it never aborts; once aborted, writes fail and flush and close do
nothing.
+ */
+ static final class DeadlineOutputStream extends FilterOutputStream {
+
+ // how often to look for the commit; the deadline may start this much
late
+ private static final long COMMIT_POLL_MILLIS = 1000;
+
+ private static final int WRITING = 0;
+ private static final int DONE = 1;
+ private static final int ABORTED = 2;
+
+ private final long maxWriteMillis;
+ private final Scheduler scheduler;
+ private final BooleanSupplier committed;
+ private final Runnable abort;
+ private final Message request;
+ private final AtomicInteger state = new AtomicInteger(WRITING);
+ private volatile Scheduler.Task timer;
+
+ /**
+ * @param abort closes the connection under any blocked write
+ * @param request the inbound message, for the log line the timer
thread writes
+ */
+ DeadlineOutputStream(OutputStream out, long maxWriteMillis, Scheduler
scheduler,
+ BooleanSupplier committed, Runnable abort,
Message request) {
+ super(out);
+ this.maxWriteMillis = maxWriteMillis;
+ this.scheduler = scheduler;
+ this.committed = committed;
+ this.abort = abort;
+ this.request = request;
+ }
+
+ private OutputStream getDelegate() {
+ return out;
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ start();
+ try {
+ out.write(b);
+ } catch (IOException e) {
+ throw timedOutOr(e);
+ }
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ start();
+ try {
+ out.write(b, off, len);
+ } catch (IOException e) {
+ throw timedOutOr(e);
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ if (state.get() == ABORTED) {
+ return;
+ }
+ try {
+ out.flush();
+ } catch (IOException e) {
+ throw timedOutOr(e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (state.get() == ABORTED) {
+ return;
+ }
+ try {
+ out.close();
+ } catch (IOException e) {
+ // aborted during the final flush: already logged, and the
connection is gone
+ if (state.get() != ABORTED) {
+ throw e;
+ }
+ }
+ }
+
+ boolean isAborted() {
+ return state.get() == ABORTED;
+ }
+
+ void complete() {
+ if (state.compareAndSet(WRITING, DONE)) {
+ Scheduler.Task t = timer;
+ if (t != null) {
+ t.cancel();
+ }
+ }
+ }
+
+ private void start() throws IOException {
+ if (state.get() == ABORTED) {
+ throw timeout();
+ }
+ if (timer == null && state.get() == WRITING) {
+ awaitCommit();
+ }
+ }
+
+ private void awaitCommit() {
+ if (state.get() != WRITING) {
+ return;
+ }
+ // a task scheduled after complete() cancels the old one is
harmless: expire() CAS-fails
+ timer = committed.getAsBoolean()
+ ? scheduler.schedule(this::expire, maxWriteMillis,
TimeUnit.MILLISECONDS)
+ : scheduler.schedule(this::awaitCommit,
Math.min(COMMIT_POLL_MILLIS, maxWriteMillis),
+ TimeUnit.MILLISECONDS);
+ }
+
+ private void expire() {
+ if (state.compareAndSet(WRITING, ABORTED)) {
+ TransferLimitLog.log(request, timeout().getMessage());
+ abort.run();
+ }
+ }
+
+ private IOException timedOutOr(IOException e) {
+ return state.get() == ABORTED ? timeout() : e;
+ }
+
+ private ResponseWriteTimeoutException timeout() {
+ return new ResponseWriteTimeoutException(
+ "Response not written within maxResponseWriteMillis (" +
maxWriteMillis + ")");
Review Comment:
The timeout message prints a bare number without units. Include `ms` (or
format as a duration) so logs are self-explanatory, especially since this is
intended to be a single-line operational log.
##########
tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/MaxRequestSizeIntegrationTest.java:
##########
@@ -0,0 +1,606 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.server.standard;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.BufferedReader;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.zip.GZIPOutputStream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import jakarta.ws.rs.core.Response;
+import org.apache.cxf.jaxrs.client.WebClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.server.core.IntegrationTestBase;
+import org.apache.tika.server.core.TestPortAllocator;
+
+/**
+ * Transfer limits on a real server, across every body-accepting endpoint its
welcome page
+ * lists.
+ */
+public class MaxRequestSizeIntegrationTest extends IntegrationTestBase {
+
+ // above what /language reads, so that endpoint cannot trip the limit by
reading alone
+ private static final int MAX_BYTES = 200_000;
+ private static final int MAX_READ_MILLIS = 3_000;
+ private static final int MAX_WRITE_MILLIS = 3_000;
+ private static final int IDLE_TIMEOUT_MILLIS = 2_000;
+ // room for two maximum-size bodies and part of a third
+ private static final int MAX_IN_FLIGHT_BYTES = 3 * MAX_BYTES;
+ // past the 16 MiB CXF discards after a response, so a reply proves
reading stopped
+ private static final int SENT_BYTES = 20 * 1024 * 1024;
+ private static final String BOUNDARY = "tika-limit-boundary";
+ // from tika-server-core's test preset catalog
+ private static final String PRESET = "catalog-xml-content";
+ // text in bigTextZip(): far more than the socket buffers hold
+ private static final int TEXT_BYTES = 16 * 1024 * 1024;
+ private static final String[] TIKA_TEXT = {"PUT", "/tika/text",
"text/plain"};
+
+ private static final Map<String, byte[]> PREAMBLES = new LinkedHashMap<>();
+
+ static {
+ PREAMBLES.put("text/plain", new byte[0]);
+ // an open JSON string, so a JSON reader keeps consuming
+ PREAMBLES.put("application/json", "{\"a\":\"".getBytes(UTF_8));
+ PREAMBLES.put("multipart/form-data; boundary=" + BOUNDARY,
+ ("--" + BOUNDARY + "\r\nContent-Disposition: form-data;
name=\"file\"; "
+ + "filename=\"a.txt\"\r\nContent-Type:
text/plain\r\n\r\n").getBytes(UTF_8));
+ }
+
+ @TempDir
+ static Path classWorkingDir;
+
+ private final int metricsPort = TestPortAllocator.findFreePort();
+
+ @BeforeAll
+ public void startServer() throws Exception {
+ // every endpoint: the defaults, plus pipes/async via allowPipes and
the config routes
+ String config = """
+ {
+ "server": {
+ "allowPipes": true,
+ "allowPerRequestConfig": true,
+ "allowHttp2": true,
+ "maxRequestSizeBytes": %d,
+ "maxRequestReadMillis": %r,
+ "maxResponseWriteMillis": %w,
+ "idleTimeoutMillis": %i,
+ "maxInFlightRequestBytes": %c,
+ "metricsPort": %m
+ },
+ "pipes": { "numClients": 1, "forkedJvmArgs": ["-Xmx256m"] },
+ "plugin-roots": "%s"
+ }
+ """
+ .replace("%d", Integer.toString(MAX_BYTES))
+ .replace("%r", Integer.toString(MAX_READ_MILLIS))
+ .replace("%w", Integer.toString(MAX_WRITE_MILLIS))
+ .replace("%i", Integer.toString(IDLE_TIMEOUT_MILLIS))
+ .replace("%c", Integer.toString(MAX_IN_FLIGHT_BYTES))
+ .replace("%m", Integer.toString(metricsPort))
+ .replace("%s",
Paths.get("target/plugins").toAbsolutePath().toString().replace("\\", "/"));
+ Path configFile =
Files.writeString(classWorkingDir.resolve("request-limit.json"), config);
+ startClassProcess(new String[]{"-config", configFile.toString(),
"--preset", PRESET},
+ classWorkingDir);
+ }
+
+ /** Whatever a test did to the server, every request's share of the body
budget comes back. */
+ @AfterEach
+ public void assertBudgetDrained() throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ long inFlight;
+ do {
+ inFlight = bodyBytesInFlight();
+ if (inFlight != 0) {
+ Thread.sleep(50);
+ }
+ } while (inFlight != 0 && System.nanoTime() < deadline);
+ assertEquals(0, inFlight, "request body budget not released");
+ }
+
+ @Test
+ public void testEveryEndpointAnswersOverLimitChunkedBeforeBodyEnds()
throws Exception {
+ List<String[]> endpoints = bodyEndpoints();
+ assertTrue(endpoints.size() > 40, "welcome page listed too few
endpoints: " + endpoints.size());
+ List<String> failures = new ArrayList<>();
+ for (String[] endpoint : endpoints) {
+ boolean anyTooLarge = false;
+ List<String> statuses = new ArrayList<>();
+ for (Map.Entry<String, byte[]> e : PREAMBLES.entrySet()) {
+ String status = sendChunked(endpoint, e.getKey(),
e.getValue());
+ statuses.add(e.getKey() + "=" + status);
+ // fail fast: each further hang would cost another socket
timeout
+ assertNotNull(status, String.join(" ", endpoint) + " " +
e.getKey()
+ + ": no response before the body ended");
+ if (status.startsWith("2")) {
+ failures.add(String.join(" ", endpoint) + " " + e.getKey()
+ " -> " + status);
+ }
+ anyTooLarge |= "413".equals(status);
+ }
+ if (!anyTooLarge) {
+ failures.add(String.join(" ", endpoint) + " never reached the
limit: " + statuses);
+ }
+ }
+ assertTrue(failures.isEmpty(), String.join("\n", failures));
+ }
+
+ @Test
+ public void testEveryEndpointRejectsOverLimitContentLength() throws
Exception {
+ List<String> failures = new ArrayList<>();
+ for (String[] endpoint : bodyEndpoints()) {
+ List<String> statuses = new ArrayList<>();
+ for (Map.Entry<String, byte[]> e : PREAMBLES.entrySet()) {
+ statuses.add(sendDeclared(endpoint, e.getKey(), e.getValue()));
+ }
+ // other types may be refused (415) before the size is looked at
+ if (!statuses.contains("413") || statuses.stream().anyMatch(s -> s
== null || s.startsWith("2"))) {
+ failures.add(String.join(" ", endpoint) + " -> " + statuses);
+ }
+ }
+ assertTrue(failures.isEmpty(), String.join("\n", failures));
+ }
+
+ /** A body dripped in under the size limit is cut off at the read
deadline. */
+ @Test
+ public void testSlowBodyTimesOut() throws Exception {
+ assertWithinReadDeadline("408", sendThenDrip(TIKA_TEXT, 0));
+ }
+
+ /** CXF discards up to 16 MiB left over after a 413; a drip must not
stretch that out. */
+ @Test
+ public void testSlowRemainderAfterOverLimitTimesOut() throws Exception {
+ assertWithinReadDeadline("413", sendThenDrip(TIKA_TEXT, 2 *
MAX_BYTES));
+ }
+
+ /** Refused before any resource runs, the body is still discarded under
the read deadline. */
+ @Test
+ public void testSlowBodyToUnmatchedPathTimesOut() throws Exception {
+ assertWithinReadDeadline("404", sendThenDrip(new String[]{"PUT",
"/nonexistent", "*/*"}, 0));
+ }
+
+ /**
+ * CXF answers OPTIONS itself, before any resource or filter runs; the
body must still be
+ * discarded, not spooled to disk without limit.
+ */
+ @Test
+ public void testOptionsBodyNotSpooled() throws Exception {
+ for (String path : new String[]{"/tika", "/rmeta"}) {
+ String[] reply = sendThenDrip(new String[]{"OPTIONS", path,
"*/*"}, SENT_BYTES);
+ assertEquals("200", reply[0]);
+ // spooling reads on until the read deadline; discarding stops at
16 MiB and answers
+ assertTrue(Long.parseLong(reply[1]) < MAX_READ_MILLIS / 2,
"answered after " + reply[1] + " ms");
+ }
+ }
+
+ private static void assertWithinReadDeadline(String expected, String[]
reply) {
+ assertEquals(expected, reply[0], "after " + reply[1] + " ms");
+ // the idle timeout alone would answer only once the drip ends
+ assertTrue(Long.parseLong(reply[1]) < 2 * MAX_READ_MILLIS, "answered
after " + reply[1] + " ms");
+ }
+
+ /** Bodies held in flight fill the budget; the request that would go over
gets 429. */
+ @Test
+ public void testOverBudgetRefused() throws Exception {
+ int held = MAX_BYTES - 10_000;
+ List<Socket> uploads = new ArrayList<>();
+ try {
+ for (int i = 0; i < 3; i++) {
+ uploads.add(startUpload(held));
+ }
+ awaitBodyBytesInFlight(3L * held);
+ try (Socket socket = new Socket("localhost", integrationTestPort))
{
+ socket.setSoTimeout(10_000);
+ OutputStream out = socket.getOutputStream();
+ out.write(requestHead(TIKA_TEXT, "text/plain")
+ .append("Transfer-Encoding:
chunked\r\n\r\n").toString().getBytes(UTF_8));
+ writeChunk(out, new byte[MAX_IN_FLIGHT_BYTES - 3 * held + 1]);
+ out.write("0\r\n\r\n".getBytes(UTF_8));
+ out.flush();
+ assertEquals("429", readStatus(socket));
+ }
+ } finally {
+ for (Socket upload : uploads) {
+ upload.close();
+ }
+ }
+ }
+
+ /** Clients that send slowly hold only what they have sent, so they cannot
lock others out. */
+ @Test
+ public void testDripsDoNotExhaustBudget() throws Exception {
+ List<Socket> uploads = new ArrayList<>();
+ try {
+ for (int i = 0; i < 16; i++) {
+ uploads.add(startUpload(0));
+ }
+ awaitBodyBytesInFlight(16);
+ assertEquals("200", sendSmall());
+ } finally {
+ for (Socket upload : uploads) {
+ upload.close();
+ }
+ }
+ }
+
+ private void awaitBodyBytesInFlight(long atLeast) throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
+ while (bodyBytesInFlight() < atLeast) {
+ assertTrue(System.nanoTime() < deadline, "uploads not in flight: "
+ bodyBytesInFlight());
+ Thread.sleep(20);
+ }
+ }
+
+ private long bodyBytesInFlight() {
+ String scrape = WebClient.create("http://localhost:"; + metricsPort +
"/metrics").get()
+ .readEntity(String.class);
+ for (String line : scrape.split("\n")) {
+ if (line.startsWith("tika_server_request_body_in_flight_bytes ")) {
+ return (long) Double.parseDouble(line.substring(line.indexOf('
') + 1).trim());
+ }
+ }
+ throw new AssertionError("no body budget gauge in:\n" + scrape);
+ }
+
+ private String sendSmall() throws IOException {
+ byte[] body = "hello world".getBytes(UTF_8);
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(10_000);
+ OutputStream out = socket.getOutputStream();
+ out.write(requestHead(new String[]{"PUT", "/tika/text",
"text/plain"}, "text/plain")
+ .append("Content-Length:
").append(body.length).append("\r\n\r\n")
+ .toString().getBytes(UTF_8));
+ out.write(body);
+ out.flush();
+ return readStatus(socket);
+ }
+ }
+
+ /** An upload of {@code headBytes}, then kept in flight a byte at a time
until its socket is closed. */
+ private Socket startUpload(int headBytes) throws IOException {
+ Socket socket = new Socket("localhost", integrationTestPort);
+ OutputStream out = socket.getOutputStream();
+ Thread writer = new Thread(() -> {
+ try {
+ out.write(requestHead(TIKA_TEXT, "text/plain")
+ .append("Transfer-Encoding:
chunked\r\n\r\n").toString().getBytes(UTF_8));
+ writeChunk(out, new byte[headBytes]);
+ while (!socket.isClosed()) {
+ writeChunk(out, new byte[]{'a'});
+ out.flush();
+ Thread.sleep(100);
+ }
+ } catch (IOException | InterruptedException e) {
+ // closed
+ }
+ });
+ writer.setDaemon(true);
+ writer.start();
+ return socket;
+ }
+
+ /** A client that stops sending is dropped at the idle timeout, with the
same 408. */
+ @Test
+ public void testStalledBodyTimesOut() throws Exception {
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(5 * IDLE_TIMEOUT_MILLIS);
+ OutputStream out = socket.getOutputStream();
+ out.write(requestHead(new String[]{"PUT", "/tika/text",
"text/plain"}, "text/plain")
+ .append("Content-Length:
100\r\n\r\nabc").toString().getBytes(UTF_8));
+ out.flush();
+ assertEquals("408", readStatus(socket));
+ }
+ }
+
+ /**
+ * Sends {@code headBytes}, then a byte at a time, well inside the idle
timeout, for over
+ * three times the read deadline. Returns {status, millis from start to
status}.
+ */
+ private String[] sendThenDrip(String[] endpoint, int headBytes) throws
IOException {
+ byte[] filler = new byte[64 * 1024];
+ Arrays.fill(filler, (byte) 'a');
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(5 * MAX_READ_MILLIS);
+ OutputStream out = socket.getOutputStream();
+ long start = System.nanoTime();
+ Thread writer = new Thread(() -> {
+ try {
+ out.write(requestHead(endpoint, "text/plain")
+ .append("Transfer-Encoding:
chunked\r\n\r\n").toString().getBytes(UTF_8));
+ for (int sent = 0; sent < headBytes; sent +=
filler.length) {
+ writeChunk(out, Arrays.copyOf(filler,
Math.min(filler.length, headBytes - sent)));
+ }
+ out.flush();
+ for (int i = 0; i < 100; i++) {
+ writeChunk(out, new byte[]{'a'});
+ out.flush();
+ Thread.sleep(100);
+ }
+ } catch (IOException | InterruptedException e) {
+ // the server stopped reading and closed
+ }
+ });
+ writer.setDaemon(true);
+ writer.start();
+ String status = readStatus(socket);
+ return new String[]{status,
+
Long.toString(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start))};
+ }
+ }
+
+ private static byte[] bigTextZip() throws IOException {
+ ByteArrayOutputStream zipped = new ByteArrayOutputStream();
+ try (ZipOutputStream zip = new ZipOutputStream(zipped)) {
+ zip.putNextEntry(new ZipEntry("big.txt"));
+ byte[] line = "the quick brown fox jumps over the lazy
dog\n".getBytes(UTF_8);
+ for (int written = 0; written < TEXT_BYTES; written +=
line.length) {
+ zip.write(line);
+ }
+ }
+ assertTrue(zipped.size() < MAX_BYTES);
+ return zipped.toByteArray();
+ }
+
+ /**
+ * A client reading too slowly to finish within the write deadline is cut
off. It keeps
+ * reading, so the server's writes progress to the deadline check; the
response, from a
+ * small zip, is far larger than the socket buffers.
+ */
+ @Test
+ public void testSlowReaderCutOff() throws Exception {
+ byte[] body = bigTextZip();
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(30_000);
+ OutputStream out = socket.getOutputStream();
+ // keep-alive, so the body is chunked: a close-delimited one
cannot show a truncation
+ out.write(requestHead(new String[]{"PUT", "/tika/text",
"text/plain"}, "application/zip")
+ .append("Content-Length:
").append(body.length).append("\r\n\r\n")
+ .toString().replace("Connection: close\r\n",
"").getBytes(UTF_8));
+ out.write(body);
+ out.flush();
+ InputStream in = socket.getInputStream();
+ byte[] buffer = new byte[64 * 1024];
+ long read = 0;
+ String tail = "";
+ // about 1.3 MB/s: the whole body would take over 4 times the
deadline
+ while (true) {
+ int n;
+ try {
+ n = in.read(buffer);
+ } catch (IOException e) {
+ break;
+ }
+ if (n == -1) {
+ break;
+ }
+ read += n;
+ tail = (tail + new String(buffer, 0, n, UTF_8));
+ tail = tail.substring(Math.max(0, tail.length() - 16));
Review Comment:
This repeatedly allocates potentially large strings while only retaining the
last 16 chars, which can add avoidable GC overhead in a slow-reader test.
Consider keeping only the last few bytes/chars via a small ring buffer (bytes)
or a bounded `StringBuilder` so allocations stay small and constant.
##########
tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/ResponseWriteDeadlineInterceptorTest.java:
##########
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.server.core;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import
org.apache.tika.server.core.ResponseWriteDeadlineInterceptor.DeadlineOutputStream;
+import
org.apache.tika.server.core.ResponseWriteDeadlineInterceptor.ResponseWriteTimeoutException;
+
+/**
+ * The deadline as a timer: a write blocked by a client that reads a trickle
(each byte resetting
+ * the idle timeout) must still end at the deadline, which a check between
writes cannot do.
+ */
+public class ResponseWriteDeadlineInterceptorTest {
+
+ private static final long DEADLINE_MILLIS = 200;
+
+ private final ScheduledExecutorScheduler scheduler = new
ScheduledExecutorScheduler();
+
+ @BeforeEach
+ public void start() throws Exception {
+ scheduler.start();
+ }
+
+ @AfterEach
+ public void stop() throws Exception {
+ scheduler.stop();
+ }
+
+ @Test
+ public void testBlockedWriteAbortedAtDeadline() {
+ BlockingSink sink = new BlockingSink();
+ AtomicInteger aborts = new AtomicInteger();
+ DeadlineOutputStream out = new DeadlineOutputStream(sink,
DEADLINE_MILLIS, scheduler, () -> true, () -> {
+ aborts.incrementAndGet();
+ sink.closeConnection();
+ }, null);
+ long start = System.nanoTime();
+ assertTimeoutPreemptively(Duration.ofSeconds(10),
+ () -> assertThrows(ResponseWriteTimeoutException.class, () ->
out.write(new byte[10])));
+ assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start) >=
DEADLINE_MILLIS);
Review Comment:
The assertion `>= DEADLINE_MILLIS` with a 200ms deadline can be flaky on
loaded/slow CI due to timer granularity and scheduling jitter. Consider adding
tolerance (e.g., allow slightly earlier firing), or asserting within a range
(not too early / not unreasonably late) to keep the test stable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
