TIKA-4793: make the Pipes IPC payload limit configurable (#3009)

Adds maxIpcPayloadBytes (default 100 MiB) and a server-side
BoundedOutputStream guard so an oversized result fails as
PAYLOAD_LIMIT_EXCEEDED instead of OOMing the worker. Preserves
already-emitted statuses on overflow and fixes archive sizing.

Closes #3009

Co-authored-by: Tim Allison <tallison@apache.org>

Author: srujana-kuntumalla

CHANGES.txt

@@ -111,6 +111,11 @@ Release 4.0.0 - ???
 
   OTHER CHANGES
 
+   * PipesClient/PipesServer IPC now enforces a configurable payload limit
+     (pipes.maxIpcPayloadBytes, default 100 MB) in both directions. Results
+     that exceed the limit return PAYLOAD_LIMIT_EXCEEDED instead of causing
+     heap exhaustion; crash messages are also size-capped (TIKA-4793).
+
    * MagicDetector now compiles its regular expression once, in the
      constructor, instead of recompiling it on every match (TIKA-4796).
 

docs/modules/ROOT/pages/pipes/configuration.adoc

@@ -138,6 +138,17 @@ These settings control how parsed results are batched before sending to emitters
 |When `false`, only successfully-parsed tuples reach the emitter — files that crash, time out, or otherwise fail are dropped from the output. When `true`, every tuple is emitted, including failures (the metadata carries the exception). Turn this on if you need a complete record of what was attempted (audit, retry logic, chaos-monkey tests).
 |===
 
+== IPC Payload Limit
+
+[cols="1,1,3"]
+|===
+|Field |Default |Description
+
+|`maxIpcPayloadBytes`
+|`104857600` (100 MB)
+|Maximum size in bytes of a single IPC message between the client and the forked server. This limit is *bidirectional*: it applies both to parse results returned from the server (FINISHED) and to requests sent from the client (NEW_REQUEST). Raising it lets very large documents pass over IPC; set the forked JVM `-Xmx` to at least approximately 3× this value to keep heap usage under control. Setting it too small (below the size of a typical `FetchEmitTuple`) will cause requests to be rejected silently as `UNSPECIFIED_CRASH`. The minimum accepted value is the serialized size of a `PAYLOAD_LIMIT_EXCEEDED` response (a few dozen bytes); values below that are rejected at config load time.
+|===
+
 == Emit Strategy
 
 `emitStrategy` controls whether parsed extracts are emitted directly from the forked PipesServer or passed back to the parent process first. The default is balanced for typical workloads — tune only if you have a memory or throughput problem.

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

@@ -421,6 +421,16 @@ private PipesResult waitForServer(FetchEmitTuple t, IntermediateResult intermedi
                         if (result.emitData() instanceof EmitDataImpl emitDataImpl) {
                             emitDataImpl.setParseContext(t.getParseContext());
                         }
+                        // The server's static PAYLOAD_LIMIT_EXCEEDED fallback frame carries null
+                        // emitData/emitKey. AsyncEmitter silently skips null-emitData results, so
+                        // the document would disappear from the audit trail. Rebuild with the
+                        // original emit key using what partial metadata we have.
+                        if (result.emitData() == null
+                                && result.status() == PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED) {
+                            return buildFatalResult(t.getId(), t.getEmitKey(),
+                                    PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
+                                    intermediateResult.get());
+                        }
                         return result;
                     default:
                         throw new IOException("Unexpected message type from server: " + msg.type());

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

@@ -30,6 +30,7 @@
 import org.apache.tika.pipes.api.FetchEmitTuple;
 import org.apache.tika.pipes.api.ParseMode;
 import org.apache.tika.pipes.core.protocol.PipesMessage;
+import org.apache.tika.pipes.core.server.ServerProtocolIO;
 
 public class PipesConfig {
 
@@ -546,18 +547,26 @@ public int getMaxIpcPayloadBytes() {
     }
 
     /**
-     * Sets the maximum IPC payload size in bytes. Must be a positive value.
-     * This bounds the size of a message the client will accept back from the
-     * forked server (chiefly the FINISHED result). Request payloads
-     * (client to server) are small and use the built-in default.
+     * Sets the maximum IPC payload size in bytes. This limit is <em>bidirectional</em>:
+     * it controls both the largest result the client will accept back from the forked server
+     * (the FINISHED payload) and the largest request the server will accept from the client
+     * (the NEW_REQUEST payload). Lowering this value below the size of a typical
+     * {@link org.apache.tika.pipes.api.FetchEmitTuple} will cause requests to be rejected
+     * on the server side and reported as undiagnosable {@code UNSPECIFIED_CRASH} errors.
+     * <p>
+     * The value must be at least {@link org.apache.tika.pipes.core.server.ServerProtocolIO#MIN_FALLBACK_PAYLOAD_BYTES}
+     * so that the server can always write a {@code PAYLOAD_LIMIT_EXCEEDED} response
+     * that the client will accept.
      *
-     * @param maxIpcPayloadBytes positive payload limit in bytes
-     * @throws IllegalArgumentException if the value is not positive
+     * @param maxIpcPayloadBytes payload limit in bytes (must be &ge; {@code ServerProtocolIO.MIN_FALLBACK_PAYLOAD_BYTES})
+     * @throws IllegalArgumentException if the value is below the minimum
      */
     public void setMaxIpcPayloadBytes(int maxIpcPayloadBytes) {
-        if (maxIpcPayloadBytes <= 0) {
+        if (maxIpcPayloadBytes < ServerProtocolIO.MIN_FALLBACK_PAYLOAD_BYTES) {
             throw new IllegalArgumentException(
-                    "maxIpcPayloadBytes must be positive, got: " + maxIpcPayloadBytes);
+                    "maxIpcPayloadBytes must be at least " +
+                    ServerProtocolIO.MIN_FALLBACK_PAYLOAD_BYTES +
+                    " (minimum to carry a PAYLOAD_LIMIT_EXCEEDED response), got: " + maxIpcPayloadBytes);
         }
         this.maxIpcPayloadBytes = maxIpcPayloadBytes;
     }

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitDataImpl.java

@@ -77,13 +77,17 @@ public void setParseContext(ParseContext parseContext) {
 
     private static long estimateSizeInBytes(String id, List<Metadata> metadataList,
                                             String containerStackTrace) {
-        long sz = 36 + id.length() * 2;
-        sz += 36 + containerStackTrace.length() * 2;
+        // Estimates Java heap cost (UTF-16: 2 bytes/char + object overhead).
+        // Used by the DYNAMIC emit strategy to decide passback vs. direct-emit; it is not
+        // used to enforce the IPC payload limit (that is handled by BoundedOutputStream in
+        // ServerProtocolIO, which measures actual wire bytes during serialization).
+        long sz = 36 + id.length() * 2L;
+        sz += 36 + containerStackTrace.length() * 2L;
         for (Metadata m : metadataList) {
             for (String n : m.names()) {
-                sz += 36 + n.length() * 2;
+                sz += 36 + n.length() * 2L;
                 for (String v : m.getValues(n)) {
-                    sz += 36 + v.length() * 2;
+                    sz += 36 + v.length() * 2L;
                 }
             }
         }

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/JsonPipesIpc.java

@@ -17,6 +17,7 @@
 package org.apache.tika.pipes.core.serialization;
 
 import java.io.IOException;
+import java.io.OutputStream;
 
 import com.fasterxml.jackson.core.StreamReadConstraints;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -72,6 +73,16 @@ public static byte[] toBytes(Object obj) throws IOException {
         return OBJECT_MAPPER.writeValueAsBytes(obj);
     }
 
+    /**
+     * Serialize an object to Smile binary format, writing directly into {@code out}.
+     * Any {@link IOException} thrown by {@code out} (e.g. from a size-capped stream)
+     * propagates unchanged, letting callers distinguish payload-limit aborts from
+     * genuine I/O errors.
+     */
+    public static void toStream(Object obj, OutputStream out) throws IOException {
+        OBJECT_MAPPER.writeValue(out, obj);
+    }
+
     /**
      * Deserialize Smile binary format bytes to an object.
      */

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

@@ -104,7 +104,7 @@ public ConnectionHandler(Socket socket, SharedServerResources resources, PipesCo
         this.resources = resources;
         this.pipesConfig = pipesConfig;
         this.heartbeatIntervalMillis = pipesConfig.getHeartbeatIntervalMillis();
-        this.protocolIO = new ServerProtocolIO(input, output);
+        this.protocolIO = new ServerProtocolIO(input, output, pipesConfig.getMaxIpcPayloadBytes());
     }
 
     @Override
@@ -181,6 +181,20 @@ private void mainLoop() {
                             LOG.error("handlerId={}: config error processing request", handlerId, e);
                             handleCrash(PipesMessageType.UNSPECIFIED_CRASH, fetchEmitTuple.getId(), e);
                         } catch (Throwable t) {
+                            if (t instanceof Error) {
+                                // OOM or other JVM-level error: don't trust the heap; exit
+                                // immediately. Everything before the exit is best-effort and
+                                // inside the try -- a secondary OOM in logging or writeCrash
+                                // must not escape and leave this shared JVM alive post-Error.
+                                try {
+                                    LOG.error("handlerId={}: fatal JVM error; exiting", handlerId, t);
+                                    protocolIO.writeCrash(PipesMessageType.OOM, t);
+                                } catch (Throwable ignored) {
+                                    //swallow
+                                } finally {
+                                    System.exit(PipesMessageType.OOM.getExitCode().orElse(18));
+                                }
+                            }
                             // respond, or the client blocks until socket timeout and
                             // restarts a healthy server
                             LOG.error("handlerId={}: error processing request", handlerId, t);

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

@@ -189,7 +189,7 @@ public PipesServer(String pipesClientId, TikaLoader tikaLoader, PipesConfig pipe
         validateHeartbeatInterval(pipesConfig);
 
         emitStrategy = pipesConfig.getEmitStrategy().getType();
-        this.protocolIO = new ServerProtocolIO(input, output);
+        this.protocolIO = new ServerProtocolIO(input, output, pipesConfig.getMaxIpcPayloadBytes());
     }
 
 
@@ -402,6 +402,12 @@ public void mainLoop() {
                         try {
                             loopUntilDone(fetchEmitTuple, mergedContext, executorCompletionService, intermediateResult, countDownLatch, parseTimeout);
                         } catch (Throwable t) {
+                            if (t instanceof Error) {
+                                // OOM or other JVM-level error: exit rather than continue in a
+                                // possibly corrupt heap state.
+                                handleCrash(PipesMessageType.OOM, fetchEmitTuple.getId(), t);
+                                return; // handleCrash calls exit(); unreachable
+                            }
                             LOG.error("Serious problem processing request", t);
                         }
                         break;