TIKA-4829 -- handle byte passing more cleanly (#3043)

Author: tballison

CHANGES.txt

@@ -1,4 +1,10 @@
-Release 4.1.0 - ???
+Release 4.1.0 - unreleased
+
+   * Pipes IPC: carry inline document bytes as a raw binary field beside the
+     tuple in the request envelope -- never inside the tuple or its
+     ParseContext -- and disable Smile's 7-bit binary encoding. Tuple JSON
+     serialized by 4.0.0 with an "inline-bytes" parse-context entry no longer
+     loads; it is rejected with a tailored message (TIKA-4829).
 
    * Digesting embedded documents no longer buffers each embedded object to a
      temp file. Zip entries are re-read from the parent archive on rewind, and
@@ -34,7 +40,6 @@ Release 4.1.0 - ???
      detection but also gains the attachments. Disable via
      "raw-tiff-parser": {"extractPreviews": false} (TIKA-4824).
 
-
 Release 4.0.0 - 8/18/2026
 
   This section is the complete delta from 3.x. It includes everything first

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

@@ -70,7 +70,7 @@ IDs you configure yourself may contain only letters, digits, `.`, `_` and `-`, a
 
 A host that already holds a document -- tika-server serving `/tika`, or an application calling `PipesForkParser` with an in-memory stream -- does not have to write it to disk for the forked worker to read. Content at or below xref:pipes/configuration.adoc#payload-limits[`maxInlineBytes`] travels inside the request and is served in the worker by the built-in `\_\_bytes` fetcher; larger content is written once to a file instead. A stream that is already backed by a file always keeps its file.
 
-This is automatic and needs no configuration. `\_\_bytes` is not declarable in a config file and cannot be named by a request.
+This is automatic and needs no configuration. `\_\_bytes` is not declarable in a config file and cannot be named by a request. The payload itself has no request-suppliable form either: it travels beside the tuple in the host's parent-to-worker request envelope, never as a tuple field (`inlineBytes` in a request tuple is rejected). The IPC wire format is internal and same-version-only; parent and worker run from the same classpath by default, and a version-skewed worker classpath (a `-cp` override in `forkedJvmArgs`) is unsupported.
 
 [#plugins]
 == Available Fetchers

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

@@ -156,7 +156,11 @@ curl -X POST http://localhost:9998/async -H "Content-Type: application/json" \
 
 A tuple's fields are `id`, `fetcher`, `fetchKey`, `emitter`, `emitKey`, and optionally
 `fetchRangeStart`, `fetchRangeEnd`, `metadata`, `parse-context` and `onParseException`. Any other
-field is a `400` — there is no silent tolerance for a typo. The `{"tuples":[...]}` envelope is
+field is a `400` — there is no silent tolerance for a typo. There is no field for content:
+`inlineBytes` draws a tailored `400`, because inline content is
+xref:pipes/fetchers.adoc#reserved-ids[how the server feeds its own workers], not something a
+tuple can carry. To parse content you already hold, PUT it to `/tika` or `/rmeta`, which inline
+it for you. The `{"tuples":[...]}` envelope is
 required on `/async`; a bare array is rejected.
 
 === Best practices

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

@@ -49,6 +49,7 @@
 import org.apache.tika.pipes.core.protocol.PipesMessage;
 import org.apache.tika.pipes.core.protocol.PipesMessageType;
 import org.apache.tika.pipes.core.serialization.JsonPipesIpc;
+import org.apache.tika.pipes.core.serialization.PipesRequest;
 import org.apache.tika.pipes.core.server.IntermediateResult;
 import org.apache.tika.utils.ExceptionUtils;
 import org.apache.tika.utils.StringUtils;
@@ -240,6 +241,14 @@ public PipesResult process(FetchEmitTuple t) throws IOException, InterruptedExce
             serverManager.connectionAbandoned();
             closeConnection();
             throw e;
+        } catch (PayloadLimitExceededException e) {
+            // Only writeTask's pre-send check throws this here (waitForServer handles its
+            // own); nothing was written, so the connection stays in sync -- keep it.
+            LOG.warn("clientId={}: request too large for id={}: {}", pipesClientId, t.getId(),
+                    e.getMessage());
+            return buildFatalResult(t.getId(), t.getEmitKey(),
+                    PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
+                    intermediateResult.get(), e.getMessage());
         } catch (Exception e) {
             LOG.error("exception waiting for server to complete task: {} ", t.getId(), e);
             closeConnection();
@@ -340,7 +349,14 @@ private void writeTask(FetchEmitTuple t) throws IOException {
             throw new IOException("connection closed");
         }
         LOG.debug("pipesClientId={}: sending NEW_REQUEST for id={}", pipesClientId, t.getId());
-        byte[] bytes = JsonPipesIpc.toBytes(t);
+        byte[] bytes = JsonPipesIpc.toBytes(PipesRequest.of(t));
+        // Fail fast before sending: the server would refuse the frame anyway, but only by
+        // dying or dropping the connection, misreported as a crash.
+        if (bytes.length > maxIpcPayloadBytes) {
+            throw new PayloadLimitExceededException("serialized request for id=" + t.getId()
+                    + " is " + bytes.length + " bytes, over maxIpcPayloadBytes="
+                    + maxIpcPayloadBytes + "; raise maxIpcPayloadBytes or shrink the request");
+        }
         PipesMessage.newRequest(bytes).write(tuple.output);
     }
 

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

@@ -601,7 +601,7 @@ public void setMaxInlineBytes(int maxInlineBytes) {
      * (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.
+     * client-side before sending, reported as {@code PAYLOAD_LIMIT_EXCEEDED}.
      * <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

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/BytesFetcher.java

@@ -29,7 +29,8 @@
 import org.apache.tika.utils.StringUtils;
 
 /**
- * Serves the bytes a caller put in the {@link InlineBytes} parse-context entry, so a host that
+ * Serves the bytes in the {@link InlineBytes} parse-context entry -- set by the caller
+ * in-process, or planted by the server from the {@code PipesRequest} envelope -- so a host that
  * already holds the content does not have to spool it to disk purely to hand it across the
  * process boundary.
  * <p>

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/InlineBytes.java

@@ -19,18 +19,19 @@
 import java.io.Serializable;
 import java.util.Arrays;
 
-import org.apache.tika.annotation.TikaComponent;
-
 /**
  * Document bytes carried in the {@code ParseContext} instead of fetched from a source, for
  * callers that already hold the content and would otherwise have to spool it to disk just to
  * hand it to the forked worker.
  * <p>
  * Read by {@link BytesFetcher}, which the tuple selects with fetcher id
- * {@link BytesFetcher#FETCHER_ID}. The IPC is Smile, so this rides as native binary rather than
- * base64; it counts against {@code maxIpcPayloadBytes} like any other part of the request.
+ * {@link BytesFetcher#FETCHER_ID}. In-process only: deliberately not a registered component,
+ * so no serialized form of it exists and serialization refuses loudly. On the IPC wire the
+ * payload travels beside
+ * the tuple in {@code PipesRequest} (which lifts it out on the parent and plants it back into
+ * the worker's context on the child); it counts against {@code maxIpcPayloadBytes} like any
+ * other part of the request. A request can supply it in no form at all.
  */
-@TikaComponent(name = "inline-bytes", spi = false)
 public class InlineBytes implements Serializable {
 
     private static final long serialVersionUID = 1L;

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PayloadLimitExceededException.java

@@ -19,13 +19,14 @@
 import java.io.IOException;
 
 /**
- * Thrown when an incoming IPC payload's declared length exceeds the configured limit
+ * Thrown when an IPC payload exceeds the configured limit
  * (see {@link org.apache.tika.pipes.core.PipesConfig#getMaxIpcPayloadBytes()};
- * default {@link PipesMessage#MAX_PAYLOAD_BYTES}). The payload bytes were not consumed,
- * so the stream is desynchronized and the connection must be closed. With a shared server
- * the process keeps running (only this connection ends); with the default per-client forked
- * server the process may still exit on the failed write, and the client reconnects on the
- * next task.
+ * default {@link PipesMessage#MAX_PAYLOAD_BYTES}). On the read side the payload bytes were
+ * not consumed, so the stream is desynchronized and the connection must be closed: with a
+ * shared server the process keeps running (only this connection ends); with the default
+ * per-client forked server the process may still exit on the failed write, and the client
+ * reconnects on the next task. On the send side ({@code PipesClient} pre-send check) nothing
+ * was written and the connection stays usable.
  */
 public class PayloadLimitExceededException extends IOException {
     public PayloadLimitExceededException(String message) {

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

@@ -32,6 +32,7 @@
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
+import java.util.TreeSet;
 
 import com.fasterxml.jackson.core.JacksonException;
 import com.fasterxml.jackson.core.JsonParser;
@@ -49,6 +50,9 @@
 
 public class FetchEmitTupleDeserializer extends JsonDeserializer<FetchEmitTuple> {
 
+    /** The parse-context name InlineBytes was registered under in 4.0.0. */
+    static final String LEGACY_INLINE_BYTES_ENTRY = "inline-bytes";
+
     private static final Set<String> KNOWN_KEYS = Set.of(
             ID, FETCHER, FETCH_KEY, EMITTER, EMIT_KEY, FETCH_RANGE_START, FETCH_RANGE_END,
             METADATA_KEY, PARSE_CONTEXT, ON_PARSE_EXCEPTION);
@@ -80,6 +84,20 @@ public static FetchEmitTupleDeserializer internal() {
     @Override
     public FetchEmitTuple deserialize(JsonParser jsonParser, DeserializationContext deserializationContext) throws IOException, JacksonException {
         JsonNode root = jsonParser.readValueAsTree();
+        // Both checked before rejectUnknownKeys so they get tailored messages.
+        if (root.has(PipesRequest.INLINE_BYTES)) {
+            throw new IOException("'" + PipesRequest.INLINE_BYTES
+                    + "' is not a FetchEmitTuple field; content travels outside the tuple, and"
+                    + " only on the host's internal IPC. For tika-server, PUT content you"
+                    + " already hold to /tika or /rmeta, which inline it for you.");
+        }
+        if (root.path(PARSE_CONTEXT).has(LEGACY_INLINE_BYTES_ENTRY)) {
+            // 4.0.0 serialized this entry; the generic "check for a typo" would mislead upgraders.
+            throw new IOException("'" + LEGACY_INLINE_BYTES_ENTRY + "' is no longer a serializable"
+                    + " parse-context entry (4.0.0 wrote it as base64): content travels outside"
+                    + " the tuple, and only on the host's internal IPC. For tika-server, PUT"
+                    + " content you already hold to /tika or /rmeta, which inline it for you.");
+        }
         rejectUnknownKeys(root);
 
         String id = readVal(ID, root, null, true);
@@ -124,7 +142,7 @@ private static void rejectUnknownKeys(JsonNode root) throws IOException {
             String name = it.next();
             if (!KNOWN_KEYS.contains(name)) {
                 throw new IOException("Unrecognized FetchEmitTuple field '" + name
-                        + "'. Check for a typo; known fields are " + KNOWN_KEYS + ".");
+                        + "'. Check for a typo; known fields are " + new TreeSet<>(KNOWN_KEYS) + ".");
             }
         }
     }

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

@@ -25,7 +25,9 @@
 import com.fasterxml.jackson.databind.JsonSerializer;
 import com.fasterxml.jackson.databind.SerializerProvider;
 
+import org.apache.tika.parser.ParseContext;
 import org.apache.tika.pipes.api.FetchEmitTuple;
+import org.apache.tika.pipes.core.fetcher.InlineBytes;
 import org.apache.tika.utils.StringUtils;
 
 public class FetchEmitTupleSerializer extends JsonSerializer<FetchEmitTuple> {
@@ -57,8 +59,17 @@ public void serialize(FetchEmitTuple t, JsonGenerator jsonGenerator, SerializerP
             jsonGenerator.writeObjectField(METADATA_KEY, t.getMetadata());
         }
         jsonGenerator.writeStringField(ON_PARSE_EXCEPTION, t.getOnParseException().name().toLowerCase(Locale.US));
-        if (!t.getParseContext().isEmpty()) {
-            jsonGenerator.writeObjectField(PARSE_CONTEXT, t.getParseContext());
+        ParseContext parseContext = t.getParseContext();
+        // Tailored: ParseContextSerializer's generic refusal suggests registering the
+        // component -- for InlineBytes, exactly the forbidden fix.
+        if (parseContext.get(InlineBytes.class) != null) {
+            throw new IOException("A FetchEmitTuple whose ParseContext holds InlineBytes has no"
+                    + " serialized form: inline content is in-process only and, on the pipes IPC,"
+                    + " travels beside the tuple in the request envelope. Remove the InlineBytes"
+                    + " entry before serializing.");
+        }
+        if (!parseContext.isEmpty()) {
+            jsonGenerator.writeObjectField(PARSE_CONTEXT, parseContext);
         }
         jsonGenerator.writeEndObject();
     }