Add generated payload limit validator

temporal-sdk/src/payloadVisitorGenerator/java/io/temporal/internal/payload/visitor/gen/PayloadVisitorGenerator.java

@@ -163,6 +163,8 @@ static boolean isTemporal(Descriptor d) {
   }
 
   // --- Java naming ---
+  // These helpers are duplicated in temporal-serviceclient as
+  // io.temporal.internal.payload.limits.gen.ProtoNames; see ProtoClosure for why.
 
   /** Mirrors protoc's UnderscoresToCamelCase used to derive Java accessor names. */
   static String camel(String input, boolean capNext) {

temporal-sdk/src/payloadVisitorGenerator/java/io/temporal/internal/payload/visitor/gen/ProtoClosure.java

@@ -15,8 +15,22 @@
 import java.util.Set;
 
 /**
- * Shared proto-descriptor model for the build-time generators: the message closure reachable from a
+ * Proto-descriptor model for {@link PayloadVisitorGenerator}: the message closure reachable from a
  * set of seed services, and which of those messages can transitively contain a {@code Payload}.
+ *
+ * <p><b>A near-copy of this class exists in {@code temporal-serviceclient}</b>, as {@code
+ * io.temporal.internal.payload.limits.gen.ProtoClosure}, serving the payload-limits generator. The
+ * duplication is deliberate: the two generators live in different modules (the limits validator has
+ * to sit next to the gRPC interceptor in {@code temporal-serviceclient}, whose protos its own build
+ * generates), and sharing one copy would need cross-project build wiring this repo has no other
+ * instance of. The copies differ only in how each obtains its {@code Descriptor}s — compiled proto
+ * classes here, a descriptor set file there.
+ *
+ * <p><b>Payload reachability must stay identical in both.</b> A change here that narrows
+ * reachability would silently make the visitor traverse fewer messages, which is how payloads
+ * quietly stop being offloaded; the same change there surfaces loudly as an unclassified-field
+ * build failure. Keep the two in sync, and prefer verifying a change by confirming both generators'
+ * output is byte-identical before and after it.
  */
 final class ProtoClosure {
 

temporal-serviceclient/build.gradle

@@ -75,6 +75,10 @@ sourceSets {
   }
 }
 
+// Descriptor set emitted by generateProto and read by the payload-limits generator; see the
+// payload limit validation section below.
+def apiDescriptorSet = layout.buildDirectory.file('generated/descriptors/temporal-api.desc')
+
 protobuf {
     // version/variables substitution is not supported in protobuf section.
     // protoc and protoc-gen-grpc-java versions are selected to be compatible
@@ -91,7 +95,16 @@ protobuf {
         }
     }
     generateProtoTasks {
-        all().each { task -> task.dependsOn updateSubmodules }
+        all().each { task ->
+            task.dependsOn updateSubmodules
+            // The payload-limits generator reads the API descriptors from this file rather than
+            // from compiled proto classes; see the payload limit validation section below.
+            if (task.sourceSet.name == 'main') {
+                task.generateDescriptorSet = true
+                task.descriptorSetOptions.includeImports = true
+                task.descriptorSetOptions.path = apiDescriptorSet.get().asFile.absolutePath
+            }
+        }
         all()*.builtins {
             java {
                 option 'annotate_code'
@@ -113,6 +126,48 @@ protobuf {
     generatedFilesBaseDir = "$buildDir/generated"
 }
 
+// --- Payload limit validation code generation ---
+// A build-time generator emits GeneratedPayloadLimitValidator.java, which checks the payload and
+// memo fields of outbound requests against the size limits the server enforces. It classifies every
+// payload-bearing field against hand-authored tables and fails the build if a field is unclassified
+// or a table entry is stale, so a proto change cannot land without an explicit decision.
+//
+// The generator reads the API descriptors from the descriptor set emitted by generateProto, not from
+// compiled proto classes. The protos are generated into this module's own main source set, so a
+// generator that needed them compiled would form a cycle with the compilation that consumes its
+// output. Its source set therefore depends only on protobuf-java.
+sourceSets {
+    payloadLimitsGenerator {
+        java {
+            srcDirs = ['src/payloadLimitsGenerator/java']
+        }
+    }
+}
+
+dependencies {
+    payloadLimitsGeneratorImplementation "com.google.protobuf:protobuf-java:$protoVersion"
+}
+
+def generatedPayloadLimitsDir = layout.buildDirectory.dir('generated/payloadlimits/java')
+
+def generatePayloadLimitValidator = tasks.register('generatePayloadLimitValidator', JavaExec) {
+    dependsOn 'compilePayloadLimitsGeneratorJava', 'generateProto'
+    classpath = sourceSets.payloadLimitsGenerator.runtimeClasspath
+    mainClass = 'io.temporal.internal.payload.limits.gen.PayloadLimitValidatorGenerator'
+    args apiDescriptorSet.get().asFile.absolutePath, generatedPayloadLimitsDir.get().asFile.absolutePath
+    inputs.file(apiDescriptorSet)
+    inputs.files(sourceSets.payloadLimitsGenerator.runtimeClasspath)
+    outputs.dir(generatedPayloadLimitsDir)
+}
+
+sourceSets.main.java.srcDir(generatePayloadLimitValidator)
+
+tasks.named('compilePayloadLimitsGeneratorJava') {
+    options.encoding = 'UTF-8'
+    options.compilerArgs << '-Xlint:none' << '-Xlint:deprecation' << '-Werror'
+    options.errorprone.error('MissingCasesInEnumSwitch')
+}
+
 javadocJar {
     dependsOn 'generateProto'
     from(file("$buildDir/generated/main/java"))

temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/CollectingSink.java

@@ -0,0 +1,65 @@
+package io.temporal.internal.payload.limits;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A {@link PayloadLimitSink} that collects violations without logging or policy decisions.
+ *
+ * <p>Each checked field is sorted into {@link #getWarnings()} or {@link #getErrors()}: a field over
+ * its error threshold (when {@code enforceError} and an error threshold are set) is an error;
+ * otherwise a field over its warning threshold is a warning.
+ */
+final class CollectingSink implements PayloadLimitSink {
+  private final PayloadLimits limits;
+  private final PayloadPath path = new PayloadPath();
+  private final List<PayloadLimitViolation> warnings = new ArrayList<>();
+  private final List<PayloadLimitViolation> errors = new ArrayList<>();
+
+  CollectingSink(PayloadLimits limits) {
+    this.limits = limits;
+  }
+
+  @Override
+  public void check(String fieldName, LimitClass limitClass, long size, boolean enforceError) {
+    long error = limits.error(limitClass);
+    long warn = limits.warn(limitClass);
+    if (enforceError && error > 0 && size > error) {
+      errors.add(
+          new PayloadLimitViolation(
+              path.leaf(fieldName), limitClass, LimitSeverity.ERROR, size, error));
+    } else if (warn > 0 && size > warn) {
+      warnings.add(
+          new PayloadLimitViolation(
+              path.leaf(fieldName), limitClass, LimitSeverity.WARNING, size, warn));
+    }
+  }
+
+  @Override
+  public void enter(String name) {
+    path.push(name);
+  }
+
+  @Override
+  public void enter(String name, int index) {
+    path.push(name, index);
+  }
+
+  @Override
+  public void enter(String name, String key) {
+    path.push(name, key);
+  }
+
+  @Override
+  public void exit() {
+    path.pop();
+  }
+
+  List<PayloadLimitViolation> getWarnings() {
+    return warnings;
+  }
+
+  List<PayloadLimitViolation> getErrors() {
+    return errors;
+  }
+}

temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitClass.java

@@ -0,0 +1,9 @@
+package io.temporal.internal.payload.limits;
+
+/** Which server-enforced size limit a payload field is subject to. */
+enum LimitClass {
+  /** Subject to the blob (payload) size limit. */
+  BLOB,
+  /** Subject to the memo size limit. */
+  MEMO
+}

temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitSeverity.java

@@ -0,0 +1,9 @@
+package io.temporal.internal.payload.limits;
+
+/** Whether a violation exceeded the warning threshold or the error threshold. */
+enum LimitSeverity {
+  /** Exceeded the warning threshold. */
+  WARNING,
+  /** Exceeded the error threshold. */
+  ERROR
+}

temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitSink.java

@@ -0,0 +1,33 @@
+package io.temporal.internal.payload.limits;
+
+/**
+ * Receives one callback per validated payload field, with the field's size as the server measures
+ * it. Implementors decide how to handle warnings and errors.
+ *
+ * <p>The generated traversal ({@code GeneratedPayloadLimitValidator}) calls {@link #enter}/{@link
+ * #exit} around each nested message so the sink can track a field's location for {@link #check}.
+ */
+interface PayloadLimitSink {
+  /**
+   * Called for each validated payload field.
+   *
+   * @param fieldName the leaf field's proto name
+   * @param limitClass which limit the field is subject to
+   * @param size the field's size in bytes for the given class
+   * @param enforceError when {@code false}, the field may warn but must never produce an
+   *     error-level violation
+   */
+  void check(String fieldName, LimitClass limitClass, long size, boolean enforceError);
+
+  /** Enter a singular nested-message field with proto name {@code name}. */
+  void enter(String name);
+
+  /** Enter element {@code index} of a repeated nested-message field {@code name}. */
+  void enter(String name, int index);
+
+  /** Enter the entry under {@code key} of a map-valued nested-message field {@code name}. */
+  void enter(String name, String key);
+
+  /** Leave the most recently entered nested field. */
+  void exit();
+}

temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitSizes.java

@@ -0,0 +1,67 @@
+package io.temporal.internal.payload.limits;
+
+import com.google.protobuf.MessageLite;
+import io.temporal.api.common.v1.Payload;
+import io.temporal.api.common.v1.Payloads;
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * Field-size measurements, mirroring how the Temporal server measures each field. Size is the
+ * serialized proto size ({@link MessageLite#getSerializedSize()}) except for the map-aggregate
+ * helpers, which mirror the server's {@code sum(len(key) + ...)} accounting. Called from the
+ * generated {@code GeneratedPayloadLimitValidator}.
+ */
+final class PayloadLimitSizes {
+  private PayloadLimitSizes() {}
+
+  /**
+   * Serialized proto size of a single message (e.g. Payload, Payloads, Memo, or a whole Failure).
+   */
+  static long serializedSize(MessageLite message) {
+    return message.getSerializedSize();
+  }
+
+  /**
+   * Sum of serialized proto sizes over a collection of messages (e.g. repeated Payload/Failure).
+   */
+  static long serializedSizeSum(Collection<? extends MessageLite> messages) {
+    long total = 0;
+    for (MessageLite m : messages) {
+      total += m.getSerializedSize();
+    }
+    return total;
+  }
+
+  /**
+   * Aggregate size of a marker-style {@code map<string, Payloads>}, mirroring the server's {@code
+   * sum(len(key) + payloads.Size())} accounting (e.g. {@code
+   * RecordMarkerCommandAttributes.details}).
+   */
+  static long mapPayloadsSum(Map<String, Payloads> entries) {
+    long total = 0;
+    for (Map.Entry<String, Payloads> e : entries.entrySet()) {
+      total += utf8Length(e.getKey()) + e.getValue().getSerializedSize();
+    }
+    return total;
+  }
+
+  /**
+   * Aggregate size of a search-attribute/memo-style {@code map<string, Payload>}, mirroring the
+   * server's {@code sum(len(key) + len(payload.data))} accounting — note the server counts the
+   * <b>raw data</b> length here, not the serialized payload size (e.g. {@code
+   * UpsertWorkflowSearchAttributes.indexed_fields}).
+   */
+  static long mapPayloadDataSum(Map<String, Payload> entries) {
+    long total = 0;
+    for (Map.Entry<String, Payload> e : entries.entrySet()) {
+      total += utf8Length(e.getKey()) + e.getValue().getData().size();
+    }
+    return total;
+  }
+
+  private static int utf8Length(String s) {
+    return s.getBytes(StandardCharsets.UTF_8).length;
+  }
+}

temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitValidator.java

@@ -0,0 +1,56 @@
+package io.temporal.internal.payload.limits;
+
+import com.google.protobuf.Message;
+import java.util.List;
+import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Validates an outbound request message's payload/memo fields against a set of {@link
+ * PayloadLimits}, mirroring the size checks the Temporal server enforces.
+ *
+ * <p>The per-(message, field) policy is generated from the proto descriptors against the
+ * hand-authored {@code *_FIELDS} tables in {@code PayloadLimitValidatorGenerator}; adding or
+ * removing a payload-bearing field fails the build until the tables are updated.
+ */
+final class PayloadLimitValidator {
+  private static final Logger log = LoggerFactory.getLogger(PayloadLimitValidator.class);
+
+  private PayloadLimitValidator() {}
+
+  /**
+   * Validates {@code request} against {@code limits}.
+   *
+   * <p>If any field exceeded its error threshold, logs the error(s) and returns the first one
+   * without logging warnings; otherwise logs each warning and returns empty. With no error
+   * thresholds set (or a request type that carries no validated payload fields), this only warns
+   * and always returns empty.
+   */
+  static Optional<PayloadLimitViolation> validate(Message request, PayloadLimits limits) {
+    CollectingSink sink = new CollectingSink(limits);
+    GeneratedPayloadLimitValidator.dispatch(sink, request);
+
+    List<PayloadLimitViolation> errors = sink.getErrors();
+    if (!errors.isEmpty()) {
+      for (PayloadLimitViolation e : errors) {
+        log.error(
+            "{} (size={}, limit={}, path={})",
+            e.getMessage(),
+            e.getSize(),
+            e.getLimit(),
+            e.getPath());
+      }
+      return Optional.of(errors.get(0));
+    }
+    for (PayloadLimitViolation w : sink.getWarnings()) {
+      log.warn(
+          "{} (size={}, limit={}, path={})",
+          w.getMessage(),
+          w.getSize(),
+          w.getLimit(),
+          w.getPath());
+    }
+    return Optional.empty();
+  }
+}