TIKA-4809 stage 6 (#3006)

* TIKA-4809: Add maxRequestSizeBytes

* TIKA-4809: Bound the request path -- spool lifetime, fork heap, temp-file suffix

* TIKA-4809: Derive numClients from cores when unset

Author: tballison

docs/modules/ROOT/pages/pipes/cpu-sizing.adoc

@@ -166,6 +166,35 @@ report `autoCap=user-set in forkedJvmArgs`.
 [#heap-per-worker]
 == Heap per worker — rule of thumb
 
+Heap is auto-sized the same way CPU is. Left to itself, every forked JVM takes its own
+default max heap — a fixed fraction of host or container memory — so `numClients` forks have
+a combined ceiling well above what the host actually has. When `numClients > 1` and you have
+not set `-Xmx` (or `-XX:MaxRAMPercentage`/`-XX:MaxRAMFraction`) yourself, Tika injects
+`-XX:MaxRAMPercentage=75/numClients`, leaving the remainder for the parent JVM and the OS.
+The `pipes-cpu-sizing` summary line reports the decision as `heap=...`.
+
+[IMPORTANT]
+====
+**`numClients` is sized against CPU, not memory.** The rule above —
+`numClients × 2 + 2 ≤ hostCores` — considers cores only. Memory is then divided among
+however many workers that produced. On a host with many cores relative to its RAM, a
+`numClients` that is correct for CPU can leave each fork with too little heap to parse
+reliably.
+
+Tika cannot reconcile the two automatically: the parent sizes forks as a *percentage* of
+memory and has no portable way to resolve that to bytes. Each forked JVM therefore checks
+its own heap at startup and logs a `WARN` if it came up under 256 MB — the point below which
+ordinary documents, not just pathological ones, begin to fail. If you see that warning,
+lower `numClients`, raise the container memory limit, or set `-Xmx` explicitly.
+
+Cross-check both constraints yourself when sizing: `numClients × 2 + 2 ≤ hostCores` **and**
+`numClients × per-worker-heap ≤ 75% of memory`.
+====
+
+Set `-Xmx` explicitly when you know your workload: the auto-slice is a safe default, not a
+tuned one, and a fork that legitimately needs more than its slice will OOM where an untuned
+JVM might have grown into spare memory.
+
 A reasonable starting point is **~2 GB of heap per forked worker** (passed via `-Xmx2g` in `forkedJvmArgs`). The number falls out of three independent constraints any of which can dominate:
 
 * **Worst-case PDF parsing.** A handful of pathological PDFs in any reasonably large corpus will allocate hundreds of MB of intermediate object data per document — large image streams, deeply nested form fields, big embedded fonts. Smaller heaps OOM on those documents; larger heaps just let GC clean up between docs.

docs/modules/ROOT/pages/using-tika/server/index.adoc

@@ -308,6 +308,10 @@ Server behavior beyond host/port is controlled by a JSON config file passed via
 |`false`
 |Include parser stack traces in error responses. Useful in dev, dangerous in production (leaks internals).
 
+|`maxRequestSizeBytes`
+|`-1` (no limit)
+|Maximum request body in bytes; larger requests are rejected with `413`. Enforced for chunked uploads too, not just those declaring a `Content-Length`. Uploads are spooled to disk, so leaving this unset lets a caller fill the temp directory.
+
 |`logLevel`
 |_inherited_
 |`debug` or `info` to override the runtime log level.

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java

@@ -65,6 +65,22 @@ public class PerClientServerManager implements ServerManager {
      *  formula could otherwise produce slice=1. */
     private static final int MIN_AUTO_CAP_SLICE = 2;
 
+    /** Share of host/container memory the forks may collectively claim; the remainder is
+     *  left for the parent JVM, the OS, and page cache for spooled input. */
+    private static final int FORK_HEAP_BUDGET_PERCENT = 75;
+
+
+    private static boolean userSetHeap(List<String> args) {
+        return args.stream().anyMatch(a -> a.startsWith("-Xmx")
+                || a.startsWith("-XX:MaxRAMPercentage")
+                || a.startsWith("-XX:MaxRAMFraction"));
+    }
+
+    private static int forkHeapPercentage(int numClients) {
+        return Math.max(1, FORK_HEAP_BUDGET_PERCENT / numClients);
+    }
+
+
     private final PipesConfig pipesConfig;
     private final Path tikaConfigPath;
     private final int clientId;
@@ -134,8 +150,17 @@ private void logCpuSizing() {
                     ? "slice=" + slice
                     : "skipped (slice<" + MIN_AUTO_CAP_SLICE + ")";
         }
+        String heapDecision;
+        if (userSetHeap(pipesConfig.getForkedJvmArgs())) {
+            heapDecision = "user-set in forkedJvmArgs";
+        } else if (numClients <= 1) {
+            heapDecision = "n/a (single fork; JVM default)";
+        } else {
+            heapDecision = "MaxRAMPercentage=" + forkHeapPercentage(numClients);
+        }
         LOG.info("pipes-cpu-sizing: hostCores={}, numClients={}, parentReserved={}, " +
-                "autoCap={}", hostCores, numClients, PARENT_RESERVED_CORES, capDecision);
+                "autoCap={}, heap={}", hostCores, numClients, PARENT_RESERVED_CORES,
+                capDecision, heapDecision);
     }
 
     @Override
@@ -442,6 +467,19 @@ private String[] getCommandline() throws IOException {
                     .toAbsolutePath());
         }
 
+        // Heap gets the same treatment as CPU. Left alone, every fork independently
+        // takes the JVM's own default max heap (a fixed fraction of host/container
+        // memory), so numClients forks have a combined ceiling well above what the
+        // host has -- the same "each JVM thinks it owns the machine" problem the
+        // ActiveProcessorCount cap solves. Give each fork a slice of a fixed budget
+        // instead, leaving the remainder for the parent and the OS.
+        if (!userSetHeap(configArgs) && pipesConfig.getNumClients() > 1) {
+            int pct = forkHeapPercentage(pipesConfig.getNumClients());
+            configArgs.add("-XX:MaxRAMPercentage=" + pct);
+            LOG.debug("clientId={}: auto-injected -XX:MaxRAMPercentage={} (numClients={})",
+                    clientId, pct, pipesConfig.getNumClients());
+        }
+
         // If the user hasn't explicitly set -XX:ActiveProcessorCount, size each
         // forked JVM's view of CPUs to a fair slice of the host. Otherwise each
         // JVM defaults its GC, JIT, and common ForkJoinPool to "all cores", which

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java

@@ -32,7 +32,23 @@ public class PipesConfig {
 
     public static final long DEFAULT_SHUTDOWN_CLIENT_AFTER_MILLS = 300000;
 
-    public static final int DEFAULT_NUM_CLIENTS = 4;
+    /** Past this, worker count becomes a memory decision, and memory is not visible here. */
+    public static final int MAX_AUTO_NUM_CLIENTS = 4;
+
+    private static final int PARENT_RESERVED_CORES = 2;
+    private static final int MIN_CORES_PER_CLIENT = 2;
+
+    /**
+     * Worker count when the operator has not chosen one. CPU-derived, so the default
+     * satisfies Tika's own sizing rule on any host; a fixed 4 needs 10 cores and would
+     * warn about itself on smaller ones. Memory cannot participate -- no Java SE API
+     * exposes container memory -- so each fork checks its own heap at startup instead.
+     */
+    public static int defaultNumClients() {
+        int hostCores = Runtime.getRuntime().availableProcessors();
+        int byCores = (hostCores - PARENT_RESERVED_CORES) / MIN_CORES_PER_CLIENT;
+        return Math.max(1, Math.min(byCores, MAX_AUTO_NUM_CLIENTS));
+    }
 
     public static final int DEFAULT_MAX_FILES_PROCESSED_PER_PROCESS = 10000;
 
@@ -66,7 +82,7 @@ public class PipesConfig {
     private long heartbeatIntervalMs = DEFAULT_HEARTBEAT_INTERVAL_MS;
 
     private long shutdownClientAfterMillis = DEFAULT_SHUTDOWN_CLIENT_AFTER_MILLS;
-    private int numClients = DEFAULT_NUM_CLIENTS;
+    private int numClients = defaultNumClients();
 
     private long maxWaitForClientMillis = DEFAULT_MAX_WAIT_FOR_CLIENT_MS;
     private int maxFilesProcessedPerProcess = DEFAULT_MAX_FILES_PROCESSED_PER_PROCESS;

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java

@@ -572,8 +572,27 @@ private static void watchParentProcess() {
         LOG.info("watching parent pid {} for exit", parentPid);
     }
 
+    /** Below this, ordinary documents -- not just pathological ones -- start OOMing. */
+    private static final long MIN_USABLE_HEAP_BYTES = 256L * 1024 * 1024;
+
+    /** Checked here, not in the parent: the parent sizes forks by percentage and has no
+     *  portable way to resolve that to bytes. The child knows what it actually got. */
+    private static void checkUsableHeap() {
+        long maxHeapMb = Runtime.getRuntime().maxMemory() / (1024 * 1024);
+        LOG.info("forked JVM max heap: {} MB", maxHeapMb);
+        if (maxHeapMb < MIN_USABLE_HEAP_BYTES / (1024 * 1024)) {
+            LOG.warn("forked JVM max heap is {} MB, below the {} MB needed to parse " +
+                            "reliably. Lower pipes.numClients, raise the container memory " +
+                            "limit, or set -Xmx explicitly in forkedJvmArgs; otherwise " +
+                            "ordinary documents will fail with OOM.",
+                    maxHeapMb, MIN_USABLE_HEAP_BYTES / (1024 * 1024));
+        }
+    }
+
     protected void initializeResources() throws TikaException, IOException, SAXException {
 
+        checkUsableHeap();
+
         TikaJsonConfig tikaJsonConfig = tikaLoader.getConfig();
         TikaPluginManager tikaPluginManager = TikaPluginManager.load(tikaJsonConfig);
 

tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/MaxRequestSizeFilter.java

@@ -0,0 +1,112 @@
+/*
+ * 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.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+import jakarta.ws.rs.container.ContainerRequestContext;
+import jakarta.ws.rs.container.ContainerRequestFilter;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.ext.Provider;
+
+/**
+ * Rejects request bodies larger than {@code maxRequestSizeBytes}.
+ * <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.
+ */
+@Provider
+public class MaxRequestSizeFilter implements ContainerRequestFilter {
+
+    static final String TOO_LARGE_MESSAGE = "Request body exceeds maxRequestSizeBytes";
+
+    private final long maxRequestSizeBytes;
+
+    /**
+     * @param maxRequestSizeBytes maximum request body in bytes; negative disables the limit
+     */
+    public MaxRequestSizeFilter(long maxRequestSizeBytes) {
+        this.maxRequestSizeBytes = maxRequestSizeBytes;
+    }
+
+    @Override
+    public void filter(ContainerRequestContext requestContext) {
+        if (maxRequestSizeBytes < 0) {
+            return;
+        }
+        if (requestContext.getLength() > maxRequestSizeBytes) {
+            requestContext.abortWith(tooLarge());
+            return;
+        }
+        requestContext.setEntityStream(
+                new BoundedInputStream(requestContext.getEntityStream(), maxRequestSizeBytes));
+    }
+
+    private static Response tooLarge() {
+        return Response
+                .status(Response.Status.REQUEST_ENTITY_TOO_LARGE)
+                .entity(TOO_LARGE_MESSAGE)
+                .type(MediaType.TEXT_PLAIN)
+                .build();
+    }
+
+    /**
+     * Throws once more than {@code limit} bytes have been read. Deliberately not
+     * silent truncation: a caller that sent too much must not receive a 200 describing
+     * a prefix of their document.
+     */
+    private static final class BoundedInputStream extends FilterInputStream {
+
+        private final long limit;
+        private long count;
+
+        private BoundedInputStream(InputStream in, long limit) {
+            super(in);
+            this.limit = limit;
+        }
+
+        @Override
+        public int read() throws IOException {
+            int c = super.read();
+            if (c != -1) {
+                add(1);
+            }
+            return c;
+        }
+
+        @Override
+        public int read(byte[] b, int off, int len) throws IOException {
+            int read = super.read(b, off, len);
+            if (read > 0) {
+                add(read);
+            }
+            return read;
+        }
+
+        private void add(int n) throws IOException {
+            count += n;
+            if (count > limit) {
+                throw new IOException(TOO_LARGE_MESSAGE + " (" + limit + ")");
+            }
+        }
+    }
+}

tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerConfig.java

@@ -69,6 +69,7 @@ public class TikaServerConfig {
     private boolean allowPerRequestConfig = false;
     private String cors = "";
     private boolean returnStackTrace = false;
+    private long maxRequestSizeBytes = -1;
     private String id = UUID
             .randomUUID()
             .toString();
@@ -240,6 +241,18 @@ public void setConfigPath(String path) {
         this.configPath = Paths.get(path);
     }
 
+    /**
+     * Maximum request body in bytes. Negative (the default) means no limit; tika-server
+     * spools uploads to disk, so an unbounded value lets a caller fill the temp directory.
+     */
+    public long getMaxRequestSizeBytes() {
+        return maxRequestSizeBytes;
+    }
+
+    public void setMaxRequestSizeBytes(long maxRequestSizeBytes) {
+        this.maxRequestSizeBytes = maxRequestSizeBytes;
+    }
+
     public boolean isReturnStackTrace() {
         return returnStackTrace;
     }

tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java

@@ -345,6 +345,7 @@ private static void loadAllProviders(TikaServerConfig tikaServerConfig, ServerSt
 
         // Add ConfigEndpointSecurityFilter to gate /config endpoints
         writers.add(new ConfigEndpointSecurityFilter(tikaServerConfig.isAllowPerRequestConfig()));
+        writers.add(new MaxRequestSizeFilter(tikaServerConfig.getMaxRequestSizeBytes()));
 
         // setRequestLogLevel rejects anything but debug/info, so no validation needed here.
         TikaLoggingFilter logFilter = null;
@@ -672,7 +673,7 @@ private static Path createServerConfig(Path existingConfigPath,
         // Only set default pipes config if there's no existing config
         // This allows user-provided config to specify their own numClients, etc.
         if (existingConfigPath == null || !Files.exists(existingConfigPath)) {
-            builder.setPipesConfig(4, null);
+            builder.setPipesConfig(PipesConfig.defaultNumClients(), null);
         }
 
         // Add unpack emitter if /unpack endpoint is enabled

tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java

@@ -190,21 +190,46 @@ public List<Metadata> parse(TikaInputStream tis, Metadata metadata,
         }
     }
 
+    /** Longest suffix carried over from a client filename; keeps well clear of NAME_MAX. */
+    private static final int MAX_SUFFIX_LENGTH = 20;
+
     /**
-     * Extracts file suffix from metadata (resource name or content-type).
+     * Extracts a file suffix from the resource name for the spool file.
+     * <p>
+     * The resource name is client-supplied ({@code Content-Disposition} / {@code File-Name}),
+     * so the suffix is sanitized here rather than left for {@code Files.createTempFile} to
+     * reject: a suffix containing a path separator makes it throw {@code IllegalArgumentException}
+     * — not a traversal, since the JDK refuses it, but an uncaught 500 driven by a request
+     * header. An over-long suffix likewise fails at the filesystem. The suffix is a parser
+     * hint, so anything unusable is simply dropped in favour of {@code .tmp}.
      */
     private String getSuffix(Metadata metadata) {
         String resourceName = metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY);
         if (resourceName != null) {
             int lastDot = resourceName.lastIndexOf('.');
             if (lastDot > 0 && lastDot < resourceName.length() - 1) {
-                return resourceName.substring(lastDot);
+                String suffix = resourceName.substring(lastDot);
+                if (isUsableSuffix(suffix)) {
+                    return suffix;
+                }
             }
         }
-        // Default suffix
         return ".tmp";
     }
 
+    private static boolean isUsableSuffix(String suffix) {
+        if (suffix.length() > MAX_SUFFIX_LENGTH) {
+            return false;
+        }
+        for (int i = 0; i < suffix.length(); i++) {
+            char c = suffix.charAt(i);
+            if (c == '/' || c == '\\' || c == '' || Character.isISOControl(c)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     /**
      * Builds a JSON error response carrying a subset of the {@code PipesResult}
      * serialization. By default the body is just {@code {"status": "TIMEOUT"}}. The