From a94b8c1a35f6a17aad8c81deda9083d24f7d362d Mon Sep 17 00:00:00 2001 From: jmaeagle99 <44687433+jmaeagle99@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:19:12 -0700 Subject: [PATCH] Add generated payload limit validator --- .../visitor/gen/PayloadVisitorGenerator.java | 2 + .../payload/visitor/gen/ProtoClosure.java | 16 +- temporal-serviceclient/build.gradle | 57 +- .../payload/limits/CollectingSink.java | 65 ++ .../internal/payload/limits/LimitClass.java | 9 + .../payload/limits/LimitSeverity.java | 9 + .../payload/limits/PayloadLimitSink.java | 33 + .../payload/limits/PayloadLimitSizes.java | 67 ++ .../payload/limits/PayloadLimitValidator.java | 56 ++ .../payload/limits/PayloadLimitViolation.java | 63 ++ .../payload/limits/PayloadLimits.java | 35 + .../internal/payload/limits/PayloadPath.java | 65 ++ .../gen/PayloadLimitValidatorGenerator.java | 867 ++++++++++++++++++ .../payload/limits/gen/ProtoClosure.java | 178 ++++ .../limits/gen/ProtoDescriptorSets.java | 96 ++ .../payload/limits/gen/ProtoNames.java | 77 ++ .../limits/PayloadLimitValidatorTest.java | 458 +++++++++ 17 files changed, 2151 insertions(+), 2 deletions(-) create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/CollectingSink.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitClass.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitSeverity.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitSink.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitSizes.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitValidator.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolation.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimits.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadPath.java create mode 100644 temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/PayloadLimitValidatorGenerator.java create mode 100644 temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoClosure.java create mode 100644 temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoDescriptorSets.java create mode 100644 temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoNames.java create mode 100644 temporal-serviceclient/src/test/java/io/temporal/internal/payload/limits/PayloadLimitValidatorTest.java diff --git a/temporal-sdk/src/payloadVisitorGenerator/java/io/temporal/internal/payload/visitor/gen/PayloadVisitorGenerator.java b/temporal-sdk/src/payloadVisitorGenerator/java/io/temporal/internal/payload/visitor/gen/PayloadVisitorGenerator.java index ad2924a51e..c2285f620e 100644 --- a/temporal-sdk/src/payloadVisitorGenerator/java/io/temporal/internal/payload/visitor/gen/PayloadVisitorGenerator.java +++ b/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) { diff --git a/temporal-sdk/src/payloadVisitorGenerator/java/io/temporal/internal/payload/visitor/gen/ProtoClosure.java b/temporal-sdk/src/payloadVisitorGenerator/java/io/temporal/internal/payload/visitor/gen/ProtoClosure.java index 9772d4b085..12bc7047e6 100644 --- a/temporal-sdk/src/payloadVisitorGenerator/java/io/temporal/internal/payload/visitor/gen/ProtoClosure.java +++ b/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}. + * + *

A near-copy of this class exists in {@code temporal-serviceclient}, 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. + * + *

Payload reachability must stay identical in both. 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 { diff --git a/temporal-serviceclient/build.gradle b/temporal-serviceclient/build.gradle index 1da1e24eee..69c087a818 100644 --- a/temporal-serviceclient/build.gradle +++ b/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")) diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/CollectingSink.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/CollectingSink.java new file mode 100644 index 0000000000..e9ae053117 --- /dev/null +++ b/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. + * + *

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 warnings = new ArrayList<>(); + private final List 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 getWarnings() { + return warnings; + } + + List getErrors() { + return errors; + } +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitClass.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitClass.java new file mode 100644 index 0000000000..515b10b2ad --- /dev/null +++ b/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 +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitSeverity.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitSeverity.java new file mode 100644 index 0000000000..f97eb421f1 --- /dev/null +++ b/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 +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitSink.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitSink.java new file mode 100644 index 0000000000..e6a581ee8f --- /dev/null +++ b/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. + * + *

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(); +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitSizes.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitSizes.java new file mode 100644 index 0000000000..d9a1bac041 --- /dev/null +++ b/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 messages) { + long total = 0; + for (MessageLite m : messages) { + total += m.getSerializedSize(); + } + return total; + } + + /** + * Aggregate size of a marker-style {@code map}, mirroring the server's {@code + * sum(len(key) + payloads.Size())} accounting (e.g. {@code + * RecordMarkerCommandAttributes.details}). + */ + static long mapPayloadsSum(Map entries) { + long total = 0; + for (Map.Entry e : entries.entrySet()) { + total += utf8Length(e.getKey()) + e.getValue().getSerializedSize(); + } + return total; + } + + /** + * Aggregate size of a search-attribute/memo-style {@code map}, mirroring the + * server's {@code sum(len(key) + len(payload.data))} accounting — note the server counts the + * raw data length here, not the serialized payload size (e.g. {@code + * UpsertWorkflowSearchAttributes.indexed_fields}). + */ + static long mapPayloadDataSum(Map entries) { + long total = 0; + for (Map.Entry 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; + } +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitValidator.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitValidator.java new file mode 100644 index 0000000000..61b7e64a63 --- /dev/null +++ b/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. + * + *

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}. + * + *

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 validate(Message request, PayloadLimits limits) { + CollectingSink sink = new CollectingSink(limits); + GeneratedPayloadLimitValidator.dispatch(sink, request); + + List 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(); + } +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolation.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolation.java new file mode 100644 index 0000000000..6c8ddf6728 --- /dev/null +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolation.java @@ -0,0 +1,63 @@ +package io.temporal.internal.payload.limits; + +/** A payload field whose size exceeded one of its configured thresholds (warning or error). */ +final class PayloadLimitViolation { + private final String path; + private final LimitClass limitClass; + private final LimitSeverity severity; + private final long size; + private final long limit; + + PayloadLimitViolation( + String path, LimitClass limitClass, LimitSeverity severity, long size, long limit) { + this.path = path; + this.limitClass = limitClass; + this.severity = severity; + this.size = size; + this.limit = limit; + } + + /** + * Path of proto field names from the root message (e.g. {@code + * commands[2].schedule_activity_task_command_attributes.input}). + */ + String getPath() { + return path; + } + + LimitClass getLimitClass() { + return limitClass; + } + + LimitSeverity getSeverity() { + return severity; + } + + /** The field's measured size in bytes. */ + long getSize() { + return size; + } + + /** + * The threshold that was exceeded (warning threshold for warnings, error threshold for errors). + */ + long getLimit() { + return limit; + } + + /** The user-facing {@code [TMPRL1103]} message. */ + String getMessage() { + String limitClassName = limitClass == LimitClass.BLOB ? "payloads" : "memo"; + String limitKind = severity == LimitSeverity.WARNING ? "warning" : "error"; + return "[TMPRL1103] Attempted to upload " + + limitClassName + + " with size that exceeded the " + + limitKind + + " limit."; + } + + @Override + public String toString() { + return getMessage(); + } +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimits.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimits.java new file mode 100644 index 0000000000..7dd01dc19e --- /dev/null +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimits.java @@ -0,0 +1,35 @@ +package io.temporal.internal.payload.limits; + +/** + * Warn/error thresholds (bytes) for both limit classes. A {@code 0} threshold disables that check + * for that class: {@code 0} warn = no warnings, {@code 0} error = no error enforcement (warnings + * only). + */ +final class PayloadLimits { + private final long blobWarn; + private final long blobError; + private final long memoWarn; + private final long memoError; + + PayloadLimits(long blobWarn, long blobError, long memoWarn, long memoError) { + this.blobWarn = blobWarn; + this.blobError = blobError; + this.memoWarn = memoWarn; + this.memoError = memoError; + } + + /** All thresholds disabled ({@code 0}). */ + static PayloadLimits none() { + return new PayloadLimits(0, 0, 0, 0); + } + + /** The warning threshold for {@code clazz}; {@code 0} means disabled. */ + long warn(LimitClass clazz) { + return clazz == LimitClass.BLOB ? blobWarn : memoWarn; + } + + /** The error threshold for {@code clazz}; {@code 0} means disabled. */ + long error(LimitClass clazz) { + return clazz == LimitClass.BLOB ? blobError : memoError; + } +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadPath.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadPath.java new file mode 100644 index 0000000000..ecfb419923 --- /dev/null +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadPath.java @@ -0,0 +1,65 @@ +package io.temporal.internal.payload.limits; + +import java.util.ArrayList; +import java.util.List; + +/** + * Path of proto field names to the field being validated; proto names keep it language-agnostic. A + * helper sinks embed to track location across {@link PayloadLimitSink#enter}/{@link + * PayloadLimitSink#exit}. + * + *

Segments are kept unrendered so that entering and leaving a message costs no allocation; the + * path string is built only by {@link #leaf}, which a sink calls only when it has a violation to + * report. The traversal runs on every outbound request, while violations are rare. + */ +final class PayloadPath { + private static final int NO_INDEX = -1; + + private final List names = new ArrayList<>(); + private final List keys = new ArrayList<>(); + private final List indexes = new ArrayList<>(); + + void push(String name) { + push(name, null, NO_INDEX); + } + + void push(String name, int index) { + push(name, null, index); + } + + void push(String name, String key) { + push(name, key, NO_INDEX); + } + + private void push(String name, String key, int index) { + names.add(name); + keys.add(key); + indexes.add(index); + } + + void pop() { + int last = names.size() - 1; + names.remove(last); + keys.remove(last); + indexes.remove(last); + } + + /** The full dotted path to a leaf field with proto name {@code fieldName}. */ + String leaf(String fieldName) { + if (names.isEmpty()) { + return fieldName; + } + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < names.size(); i++) { + sb.append(names.get(i)); + String key = keys.get(i); + if (key != null) { + sb.append('[').append(key).append(']'); + } else if (indexes.get(i) != NO_INDEX) { + sb.append('[').append(indexes.get(i)).append(']'); + } + sb.append('.'); + } + return sb.append(fieldName).toString(); + } +} diff --git a/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/PayloadLimitValidatorGenerator.java b/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/PayloadLimitValidatorGenerator.java new file mode 100644 index 0000000000..60a67295d0 --- /dev/null +++ b/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/PayloadLimitValidatorGenerator.java @@ -0,0 +1,867 @@ +package io.temporal.internal.payload.limits.gen; + +import com.google.protobuf.Descriptors.Descriptor; +import com.google.protobuf.Descriptors.DescriptorValidationException; +import com.google.protobuf.Descriptors.FieldDescriptor; +import com.google.protobuf.Descriptors.FileDescriptor; +import com.google.protobuf.Descriptors.MethodDescriptor; +import com.google.protobuf.Descriptors.ServiceDescriptor; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Deque; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.TreeSet; + +/** + * Build-time generator that emits {@code GeneratedPayloadLimitValidator}, the payload/memo + * size-limit validator mirroring the Temporal server's size checks. + * + *

Starting from the request (RPC input) messages of {@code WorkflowService} and {@code + * OperatorService}, it walks the proto closure, stopping at terminal payload/memo leaf types, and + * emits one {@code visit_*} method per payload-bearing message plus an {@code instanceof} + * dispatcher over the request roots. Each leaf is classified against the hand-authored {@code + * *_FIELDS} tables below; the per-field measurement is derived mechanically from the proto shape. + * + *

The tables are the compile-time forcing function: a payload/memo-bearing field that is missing + * from every table (or a stale/duplicate table entry) fails the build, so a proto change + * cannot land until it is explicitly classified here. + * + *

Descriptors are read from a protoc-emitted descriptor set file rather than from compiled proto + * classes, so that this generator does not depend on the compiled output of the module it generates + * into. + * + *

Usage: {@code PayloadLimitValidatorGenerator }. + */ +public final class PayloadLimitValidatorGenerator { + + static final String PAYLOAD = "temporal.api.common.v1.Payload"; + static final String PAYLOADS = "temporal.api.common.v1.Payloads"; + static final String MEMO = "temporal.api.common.v1.Memo"; + static final String HEADER = "temporal.api.common.v1.Header"; + static final String SEARCH_ATTRIBUTES = "temporal.api.common.v1.SearchAttributes"; + static final String FAILURE = "temporal.api.failure.v1.Failure"; + + static final String OUTPUT_PACKAGE = "io.temporal.internal.payload.limits"; + static final String OUTPUT_CLASS = "GeneratedPayloadLimitValidator"; + + /** + * Types the walk stops at: it emits a table-driven leaf check at the holding field rather than + * descending into their inner payload fields. {@code Failure} is measured as a whole proto + * because that is how the server size-checks it (e.g. FailWorkflowExecution). + */ + static final Set TERMINAL_LEAVES = + new HashSet<>(Arrays.asList(PAYLOAD, PAYLOADS, MEMO, HEADER, SEARCH_ATTRIBUTES, FAILURE)); + + /** + * Field paths the server size-checks as a whole serialized sub-message even though the field is + * not itself payload-bearing (so payload reachability never reaches them). Measured via whole- + * message size, classified via the table like any other leaf; the owning message is forced into + * the closure so parents recurse into it. + */ + static final String[] EXTRA_WHOLE_MESSAGE_LEAVES = { + // protocol Message body (google.protobuf.Any): the server blob-checks proto.Size(message.Body) + // when processing update messages and fails the WFT on exceed. + "temporal.api.protocol.v1.Message.body", + }; + + // =========================================================================== + // Payload-limits decision tables — the source of truth for how the SDK mirrors the server's + // payload/memo size checks. Fields are grouped by policy: + // BLOB_FIELDS / MEMO_FIELDS blob / memo limit, warn + error + // BLOB_WARN_FIELDS blob limit, warning only (enforceError = false) + // NOT_VALIDATED_FIELDS the server enforces no replicable limit on the field + // Roots are derived automatically from the seed services' RPC input messages, so a new RPC can't + // be silently missed — its payload fields become unclassified and fail the build until added + // here. + // =========================================================================== + + static final String[] BLOB_FIELDS = { + "temporal.api.command.v1.CompleteWorkflowExecutionCommandAttributes.result", + "temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes.input", + "temporal.api.command.v1.FailWorkflowExecutionCommandAttributes.failure", // whole Failure proto + "temporal.api.command.v1.ModifyWorkflowPropertiesCommandAttributes.upserted_memo", // memo + // data-sum + "temporal.api.command.v1.RecordMarkerCommandAttributes.details", // map sum + "temporal.api.command.v1.ScheduleActivityTaskCommandAttributes.input", + "temporal.api.command.v1.ScheduleNexusOperationCommandAttributes.input", + "temporal.api.command.v1.SignalExternalWorkflowExecutionCommandAttributes.input", + "temporal.api.command.v1.StartChildWorkflowExecutionCommandAttributes.input", + "temporal.api.command.v1.UpsertWorkflowSearchAttributesCommandAttributes.search_attributes", // indexed_fields data-sum + "temporal.api.protocol.v1.Message.body", // whole Any body; see EXTRA_WHOLE_MESSAGE_LEAVES + "temporal.api.query.v1.WorkflowQuery.query_args", + "temporal.api.workflow.v1.NewWorkflowExecutionInfo.input", + "temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatByIdRequest.details", + "temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatRequest.details", + "temporal.api.workflowservice.v1.RespondActivityTaskCanceledByIdRequest.details", + "temporal.api.workflowservice.v1.RespondActivityTaskCanceledRequest.details", + "temporal.api.workflowservice.v1.RespondActivityTaskCompletedByIdRequest.result", + "temporal.api.workflowservice.v1.RespondActivityTaskCompletedRequest.result", + "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest.input", + "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest.signal_input", + "temporal.api.workflowservice.v1.SignalWorkflowExecutionRequest.input", + "temporal.api.workflowservice.v1.StartActivityExecutionRequest.input", + "temporal.api.workflowservice.v1.StartNexusOperationExecutionRequest.input", + "temporal.api.workflowservice.v1.StartWorkflowExecutionRequest.input", + }; + + static final String[] MEMO_FIELDS = { + "temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes.memo", + "temporal.api.command.v1.StartChildWorkflowExecutionCommandAttributes.memo", + "temporal.api.workflow.v1.NewWorkflowExecutionInfo.memo", + "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest.memo", + "temporal.api.workflowservice.v1.StartWorkflowExecutionRequest.memo", + }; + + // Warn-only: the SDK warns but never proactively fails the task (failure responses; query + // results). + static final String[] BLOB_WARN_FIELDS = { + "temporal.api.workflowservice.v1.RespondActivityTaskFailedByIdRequest.failure", + "temporal.api.workflowservice.v1.RespondActivityTaskFailedByIdRequest.last_heartbeat_details", + "temporal.api.workflowservice.v1.RespondActivityTaskFailedRequest.failure", + "temporal.api.workflowservice.v1.RespondActivityTaskFailedRequest.last_heartbeat_details", + "temporal.api.workflowservice.v1.RespondNexusTaskFailedRequest.failure", + "temporal.api.workflowservice.v1.RespondWorkflowTaskFailedRequest.failure", + "temporal.api.query.v1.WorkflowQueryResult.answer", + "temporal.api.workflowservice.v1.RespondQueryTaskCompletedRequest.query_result", + }; + + static final String[] NOT_VALIDATED_FIELDS = { + // Headers: server records a HeaderSize metric only. + "temporal.api.batch.v1.BatchOperationSignal.header", + "temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes.header", + "temporal.api.command.v1.RecordMarkerCommandAttributes.header", + "temporal.api.command.v1.ScheduleActivityTaskCommandAttributes.header", + "temporal.api.command.v1.SignalExternalWorkflowExecutionCommandAttributes.header", + "temporal.api.command.v1.StartChildWorkflowExecutionCommandAttributes.header", + "temporal.api.query.v1.WorkflowQuery.header", + "temporal.api.update.v1.Input.header", + "temporal.api.workflow.v1.NewWorkflowExecutionInfo.header", + "temporal.api.workflow.v1.PostResetOperation.SignalWorkflow.header", + "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest.header", + "temporal.api.workflowservice.v1.SignalWorkflowExecutionRequest.header", + "temporal.api.workflowservice.v1.StartActivityExecutionRequest.header", + "temporal.api.workflowservice.v1.StartWorkflowExecutionRequest.header", + // Search attributes: separate non-replicable SA limit (server merges with existing SAs). + "temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes.search_attributes", + "temporal.api.command.v1.StartChildWorkflowExecutionCommandAttributes.search_attributes", + "temporal.api.workflow.v1.NewWorkflowExecutionInfo.search_attributes", + "temporal.api.workflowservice.v1.CreateScheduleRequest.search_attributes", + "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest.search_attributes", + "temporal.api.workflowservice.v1.StartActivityExecutionRequest.search_attributes", + "temporal.api.workflowservice.v1.StartNexusOperationExecutionRequest.search_attributes", + "temporal.api.workflowservice.v1.StartWorkflowExecutionRequest.search_attributes", + "temporal.api.workflowservice.v1.UpdateScheduleRequest.search_attributes", + // Internal carry-over fields the SDK doesn't author / the server doesn't size-check here. + "temporal.api.command.v1.CancelWorkflowExecutionCommandAttributes.details", + "temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes.failure", + "temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes.last_completion_result", + "temporal.api.command.v1.RecordMarkerCommandAttributes.failure", + "temporal.api.workflowservice.v1.StartWorkflowExecutionRequest.continued_failure", + "temporal.api.workflowservice.v1.StartWorkflowExecutionRequest.last_completion_result", + "temporal.api.workflowservice.v1.TerminateWorkflowExecutionRequest.details", + // Dedicated, non-fetchable limits: UserMetadata (nexus-start only); Nexus EndpointSpec + // description. + "temporal.api.sdk.v1.UserMetadata.details", + "temporal.api.sdk.v1.UserMetadata.summary", + "temporal.api.nexus.v1.EndpointSpec.description", + // Event group marker label: custom 400-byte server-side limit, not blob/memo. + "temporal.api.sdk.v1.EventGroupMarker.Label.label", + // Update input args: frontend records a metric only — enforced on delivery via Message.body. + "temporal.api.update.v1.Input.args", + // Query/nexus failures and the nexus sync response payload: not size-checked on these paths. + "temporal.api.nexus.v1.StartOperationResponse.Sync.payload", + "temporal.api.nexus.v1.StartOperationResponse.failure", + "temporal.api.query.v1.WorkflowQueryResult.failure", + "temporal.api.workflowservice.v1.RespondQueryTaskCompletedRequest.failure", + // Schedules: server sums memo + action.input vs blob — cross-field aggregate (deferred). + "temporal.api.workflowservice.v1.CreateScheduleRequest.memo", + "temporal.api.workflowservice.v1.UpdateScheduleRequest.memo", + // Enforced downstream: signal input is blob-checked per target on batch/reset fan-out. + "temporal.api.batch.v1.BatchOperationSignal.input", + "temporal.api.workflow.v1.PostResetOperation.SignalWorkflow.input", + // Not size-checked by the server. + "temporal.api.batch.v1.BatchOperationTermination.details", + "temporal.api.deployment.v1.UpdateDeploymentMetadata.upsert_entries", + "temporal.api.workflowservice.v1.UpdateWorkerDeploymentVersionMetadataRequest.upsert_entries", + // Cloud compute API; not size-checked by the OSS server. + "temporal.api.compute.v1.ComputeProvider.details", + "temporal.api.compute.v1.ComputeScaler.details", + }; + + // --- Policy / target model ------------------------------------------------- + + enum LimitClass { + BLOB, + MEMO; + + String token() { + return this == BLOB ? "LimitClass.BLOB" : "LimitClass.MEMO"; + } + } + + /** + * Classification of a leaf field, or {@link #NOT_VALIDATED} for a field the server doesn't check. + */ + static final class FieldPolicy { + static final FieldPolicy NOT_VALIDATED = new FieldPolicy(null, false, false); + + final boolean validated; + final LimitClass limitClass; + final boolean enforceError; + + private FieldPolicy(LimitClass limitClass, boolean enforceError, boolean validated) { + this.limitClass = limitClass; + this.enforceError = enforceError; + this.validated = validated; + } + + static FieldPolicy validated(LimitClass limitClass, boolean enforceError) { + return new FieldPolicy(limitClass, enforceError, true); + } + } + + /** How a leaf field's size is measured; derived mechanically from the proto shape. */ + enum LeafKind { + SINGLE_PAYLOADS, + SINGLE_PAYLOAD, + REPEATED_PAYLOAD, + SINGLE_MEMO, + /** A Memo validated against the blob limit: measured as the data-sum of its fields. */ + MEMO_FIELDS_DATA_SUM, + SINGLE_HEADER, + SINGLE_SEARCH_ATTRIBUTES, + MAP_PAYLOAD, + MAP_PAYLOADS, + /** A whole message measured by its serialized proto size (e.g. Failure). */ + WHOLE_MESSAGE, + /** A repeated whole message, summed. */ + REPEATED_WHOLE_MESSAGE + } + + enum StructShape { + SINGLE, + REPEATED, + MAP + } + + /** Where a field leads: a measured leaf, a recurse-into struct, or ignored. */ + static final class Target { + static final Target SKIP = new Target(null, null, null); + + final LeafKind leaf; // non-null for a leaf + final StructShape structShape; // non-null for a struct + final Descriptor child; // struct child message descriptor + + private Target(LeafKind leaf, StructShape structShape, Descriptor child) { + this.leaf = leaf; + this.structShape = structShape; + this.child = child; + } + + static Target leaf(LeafKind kind) { + return new Target(kind, null, null); + } + + static Target struct(StructShape shape, Descriptor child) { + return new Target(null, shape, child); + } + } + + // --- Entry point ----------------------------------------------------------- + + /** Proto files whose services seed the walk. */ + static final String WORKFLOW_SERVICE_PROTO = "temporal/api/workflowservice/v1/service.proto"; + + static final String OPERATOR_SERVICE_PROTO = "temporal/api/operatorservice/v1/service.proto"; + + public static void main(String[] args) throws Exception { + if (args.length < 2) { + throw new IllegalArgumentException( + "usage: PayloadLimitValidatorGenerator "); + } + new PayloadLimitValidatorGenerator().run(Paths.get(args[0]), Paths.get(args[1])); + } + + private ProtoClosure closure; + private final Map table = loadTable(); + private final Set usedKeys = new HashSet<>(); + private final Set unclassified = new TreeSet<>(); + private final Set extraLeafOwners = new HashSet<>(); + + /** + * Full names of the messages that can produce at least one check; see {@link #validatingClosure}. + */ + private Set validating = new HashSet<>(); + + void run(Path descriptorSetFile, Path outputRoot) + throws IOException, DescriptorValidationException { + Map files = ProtoDescriptorSets.load(descriptorSetFile); + List seeds = + Arrays.asList( + ProtoDescriptorSets.require(files, WORKFLOW_SERVICE_PROTO), + ProtoDescriptorSets.require(files, OPERATOR_SERVICE_PROTO)); + this.closure = ProtoClosure.of(seeds); + + for (String path : EXTRA_WHOLE_MESSAGE_LEAVES) { + int dot = path.lastIndexOf('.'); + extraLeafOwners.add(path.substring(0, dot)); + } + + // Roots: the RPC input (request) message of every method of the seed services. + Set roots = new LinkedHashSet<>(); + for (FileDescriptor seed : seeds) { + for (ServiceDescriptor service : seed.getServices()) { + for (MethodDescriptor method : service.getMethods()) { + roots.add(method.getInputType()); + } + } + } + + // Closure of payload-bearing messages reachable from the roots, keyed for deterministic output. + Map toGenerate = new TreeMap<>(); + for (Descriptor d : limitsClosure(roots)) { + toGenerate.put(d.getFullName(), d); + } + + // Classify every leaf in the closure before pruning below, so the table guards always see the + // full set of payload-bearing fields regardless of what ends up being emitted. + for (Descriptor d : toGenerate.values()) { + for (Leaf leaf : leavesOf(d)) { + leafPolicy(d.getFullName() + "." + leaf.field.getName()); + } + } + + // Fail the build on any unclassified payload-bearing field. + if (!unclassified.isEmpty()) { + StringBuilder msg = new StringBuilder(); + msg.append("payload-limits: ") + .append(unclassified.size()) + .append( + " payload-bearing field(s) are not classified. Add each to the right *_FIELDS list " + + "(BLOB_FIELDS / MEMO_FIELDS / BLOB_WARN_FIELDS / NOT_VALIDATED_FIELDS) in " + + "PayloadLimitValidatorGenerator:\n"); + for (String p : unclassified) { + msg.append(" \"").append(p).append("\",\n"); + } + throw new IllegalStateException(msg.toString()); + } + + // Fail the build on stale table entries (no longer a payload-bearing field in the closure). + Set stale = new TreeSet<>(table.keySet()); + stale.removeAll(usedKeys); + if (!stale.isEmpty()) { + StringBuilder msg = new StringBuilder(); + msg.append("payload-limits: ") + .append(stale.size()) + .append( + " stale entr(y/ies) in the *_FIELDS tables (PayloadLimitValidatorGenerator) no " + + "longer correspond to a payload-bearing field; remove them:\n"); + for (String p : stale) { + msg.append(" ").append(p).append("\n"); + } + throw new IllegalStateException(msg.toString()); + } + + // Drop the messages that can never produce a check, and with them the fields that only lead + // there. Classification above already happened, so pruning cannot weaken the build guards. + this.validating = validatingClosure(toGenerate); + toGenerate.keySet().retainAll(validating); + + String source = emit(toGenerate, roots); + + Path dir = outputRoot; + for (String part : OUTPUT_PACKAGE.split("\\.", -1)) { + dir = dir.resolve(part); + } + Files.createDirectories(dir); + Path out = dir.resolve(OUTPUT_CLASS + ".java"); + Files.write(out, source.getBytes(StandardCharsets.UTF_8)); + System.out.println( + "PayloadLimitValidatorGenerator: wrote " + toGenerate.size() + " validators to " + out); + } + + // --- Table loading --------------------------------------------------------- + + private static Map loadTable() { + Map map = new HashMap<>(); + putAll(map, BLOB_FIELDS, FieldPolicy.validated(LimitClass.BLOB, true)); + putAll(map, MEMO_FIELDS, FieldPolicy.validated(LimitClass.MEMO, true)); + putAll(map, BLOB_WARN_FIELDS, FieldPolicy.validated(LimitClass.BLOB, false)); + putAll(map, NOT_VALIDATED_FIELDS, FieldPolicy.NOT_VALIDATED); + return map; + } + + private static void putAll(Map map, String[] paths, FieldPolicy policy) { + for (String path : paths) { + if (map.put(path, policy) != null) { + throw new IllegalStateException("payload-limits: duplicate table entry for `" + path + "`"); + } + } + } + + // --- Reachability + classification ----------------------------------------- + + /** + * Whether {@code d} is part of the validated closure (payload-reachable or an extra-leaf owner). + */ + private boolean included(Descriptor d) { + return closure.reaches(d) || extraLeafOwners.contains(d.getFullName()); + } + + /** + * BFS from the roots, following payload-containing structural fields; excludes terminal leaves. + */ + private Set limitsClosure(Set roots) { + Set result = new LinkedHashSet<>(); + Set seen = new HashSet<>(); + Deque queue = new ArrayDeque<>(roots); + while (!queue.isEmpty()) { + Descriptor d = queue.poll(); + String fqn = d.getFullName(); + if (TERMINAL_LEAVES.contains(fqn) || !seen.add(fqn)) { + continue; + } + if (!included(d)) { + continue; + } + result.add(d); + for (FieldDescriptor f : d.getFields()) { + Target t = classify(f); + if (t.structShape != null && included(t.child)) { + queue.add(t.child); + } + } + } + return result; + } + + static Target classify(FieldDescriptor f) { + if (f.isMapField()) { + FieldDescriptor value = f.getMessageType().findFieldByNumber(2); + if (value.getJavaType() != FieldDescriptor.JavaType.MESSAGE) { + return Target.SKIP; + } + String name = value.getMessageType().getFullName(); + if (PAYLOAD.equals(name)) { + return Target.leaf(LeafKind.MAP_PAYLOAD); + } + if (PAYLOADS.equals(name)) { + return Target.leaf(LeafKind.MAP_PAYLOADS); + } + if (ProtoNames.isTemporal(value.getMessageType())) { + return Target.struct(StructShape.MAP, value.getMessageType()); + } + return Target.SKIP; + } + if (f.getJavaType() != FieldDescriptor.JavaType.MESSAGE) { + return Target.SKIP; + } + String name = f.getMessageType().getFullName(); + boolean repeated = f.isRepeated(); + LeafKind kind = terminalLeafKind(name, repeated); + if (kind != null) { + return Target.leaf(kind); + } + if (!ProtoNames.isTemporal(f.getMessageType())) { + return Target.SKIP; + } + return Target.struct(repeated ? StructShape.REPEATED : StructShape.SINGLE, f.getMessageType()); + } + + /** + * The leaf measurement kind for a terminal-leaf type, or {@code null} for a recurse-into message. + */ + static LeafKind terminalLeafKind(String typeName, boolean repeated) { + if (PAYLOAD.equals(typeName)) { + return repeated ? LeafKind.REPEATED_PAYLOAD : LeafKind.SINGLE_PAYLOAD; + } + if (PAYLOADS.equals(typeName)) { + return LeafKind.SINGLE_PAYLOADS; + } + if (MEMO.equals(typeName)) { + return LeafKind.SINGLE_MEMO; + } + if (HEADER.equals(typeName)) { + return LeafKind.SINGLE_HEADER; + } + if (SEARCH_ATTRIBUTES.equals(typeName)) { + return LeafKind.SINGLE_SEARCH_ATTRIBUTES; + } + if (FAILURE.equals(typeName)) { + return repeated ? LeafKind.REPEATED_WHOLE_MESSAGE : LeafKind.WHOLE_MESSAGE; + } + return null; + } + + /** A Memo validated against the blob limit is measured as its fields' data-sum. */ + static LeafKind effectiveKind(LeafKind kind, LimitClass limitClass) { + if (kind == LeafKind.SINGLE_MEMO && limitClass == LimitClass.BLOB) { + return LeafKind.MEMO_FIELDS_DATA_SUM; + } + return kind; + } + + private FieldPolicy leafPolicy(String protoPath) { + FieldPolicy policy = table.get(protoPath); + if (policy == null) { + unclassified.add(protoPath); + return null; + } + usedKeys.add(protoPath); + return policy; + } + + /** A measured leaf field of a message: the field itself, plus how its size is taken. */ + private static final class Leaf { + final FieldDescriptor field; + final LeafKind kind; + + Leaf(FieldDescriptor field, LeafKind kind) { + this.field = field; + this.kind = kind; + } + } + + /** Every measured leaf of {@code d}, in emission order: terminal leaves, then forced extras. */ + private List leavesOf(Descriptor d) { + List leaves = new ArrayList<>(); + for (FieldDescriptor f : d.getFields()) { + Target t = classify(f); + if (t.leaf != null) { + leaves.add(new Leaf(f, t.leaf)); + } + } + leaves.addAll(extraLeavesOf(d)); + return leaves; + } + + /** The {@link #EXTRA_WHOLE_MESSAGE_LEAVES} entries owned by {@code d}. */ + private List extraLeavesOf(Descriptor d) { + List leaves = new ArrayList<>(); + for (String path : EXTRA_WHOLE_MESSAGE_LEAVES) { + int dot = path.lastIndexOf('.'); + if (!path.substring(0, dot).equals(d.getFullName())) { + continue; + } + FieldDescriptor f = d.findFieldByName(path.substring(dot + 1)); + if (f != null) { + leaves.add(new Leaf(f, LeafKind.WHOLE_MESSAGE)); + } + } + return leaves; + } + + /** The struct-typed fields of {@code d} that the traversal recurses into. */ + private List structsOf(Descriptor d) { + List structs = new ArrayList<>(); + for (FieldDescriptor f : d.getFields()) { + Target t = classify(f); + if (t.structShape != null && included(t.child)) { + structs.add(t); + } + } + return structs; + } + + /** + * Messages that can produce at least one check: those with a validated leaf, plus, to a fixpoint, + * those that lead to one. Everything else traverses to nothing, so emitting it (and the fields + * that reach it) would only cost code size and per-request work. + */ + private Set validatingClosure(Map toGenerate) { + Set result = new HashSet<>(); + Map> children = new HashMap<>(); + for (Descriptor d : toGenerate.values()) { + for (Leaf leaf : leavesOf(d)) { + FieldPolicy policy = table.get(d.getFullName() + "." + leaf.field.getName()); + if (policy != null && policy.validated) { + result.add(d.getFullName()); + break; + } + } + List refs = new ArrayList<>(); + for (Target t : structsOf(d)) { + refs.add(t.child.getFullName()); + } + children.put(d.getFullName(), refs); + } + boolean changed = true; + while (changed) { + changed = false; + for (Map.Entry> e : children.entrySet()) { + if (result.contains(e.getKey())) { + continue; + } + for (String child : e.getValue()) { + if (result.contains(child)) { + result.add(e.getKey()); + changed = true; + break; + } + } + } + } + return result; + } + + // --- Emission -------------------------------------------------------------- + + private String emit(Map toGenerate, Set roots) { + StringBuilder sb = new StringBuilder(); + sb.append("// Code generated by PayloadLimitValidatorGenerator; DO NOT EDIT.\n"); + sb.append("package ").append(OUTPUT_PACKAGE).append(";\n\n"); + sb.append("import com.google.protobuf.Message;\n"); + sb.append("import java.util.HashMap;\n"); + sb.append("import java.util.List;\n"); + sb.append("import java.util.Map;\n"); + sb.append("import java.util.function.BiConsumer;\n\n"); + sb.append("@SuppressWarnings(\"deprecation\")\n"); + sb.append("final class ").append(OUTPUT_CLASS).append(" {\n"); + sb.append(" private ").append(OUTPUT_CLASS).append("() {}\n\n"); + + // Dispatcher over the payload-bearing request roots. + List rootList = new ArrayList<>(); + Set rootSeen = new HashSet<>(); + for (Descriptor r : roots) { + if (toGenerate.containsKey(r.getFullName()) && rootSeen.add(r.getFullName())) { + rootList.add(r); + } + } + rootList.sort((a, b) -> a.getFullName().compareTo(b.getFullName())); + // Keyed on the exact class: generated proto message classes are final, so there are no + // subtypes to widen for, and a single hash lookup beats a linear chain of instanceof tests -- + // above all for the requests that carry no payloads, which is most of them. + sb.append(" private static final Map, BiConsumer>") + .append(" DISPATCH = new HashMap<>();\n\n"); + sb.append(" static {\n"); + for (Descriptor r : rootList) { + String src = ProtoNames.sourceClassName(r); + sb.append(" DISPATCH.put(") + .append(src) + .append(".class, (sink, msg) -> ") + .append(ProtoNames.methodName(r.getFullName())) + .append("(sink, (") + .append(src) + .append(") msg));\n"); + } + sb.append(" }\n\n"); + sb.append(" static void dispatch(PayloadLimitSink sink, Message request) {\n"); + sb.append(" BiConsumer validator") + .append(" = DISPATCH.get(request.getClass());\n"); + sb.append(" if (validator != null) {\n"); + sb.append(" validator.accept(sink, request);\n"); + sb.append(" }\n"); + sb.append(" }\n\n"); + + for (Descriptor d : toGenerate.values()) { + emitValidateMethod(sb, d); + } + + sb.append("}\n"); + return sb.toString(); + } + + private void emitValidateMethod(StringBuilder sb, Descriptor d) { + String src = ProtoNames.sourceClassName(d); + sb.append(" static void ") + .append(ProtoNames.methodName(d.getFullName())) + .append("(PayloadLimitSink sink, ") + .append(src) + .append(" msg) {\n"); + int fi = 0; + for (FieldDescriptor f : d.getFields()) { + Target t = classify(f); + if (t.leaf != null) { + emitLeaf(sb, d.getFullName(), f, t.leaf); + } else if (t.structShape != null + && included(t.child) + && validating.contains(t.child.getFullName())) { + emitStruct(sb, f, t.structShape, t.child, fi++); + } + } + // Extra whole-message leaves whose owner is this message. + for (Leaf leaf : extraLeavesOf(d)) { + emitLeaf(sb, d.getFullName(), leaf.field, leaf.kind); + } + sb.append(" }\n\n"); + } + + private void emitLeaf(StringBuilder sb, String ownerFqn, FieldDescriptor f, LeafKind kind) { + String protoField = f.getName(); + FieldPolicy policy = leafPolicy(ownerFqn + "." + protoField); + if (policy == null || !policy.validated) { + return; // unclassified (build fails) or NotValidated (no check) + } + LeafKind effective = effectiveKind(kind, policy.limitClass); + String base = ProtoNames.base(f); + String classToken = policy.limitClass.token(); + String ee = String.valueOf(policy.enforceError); + switch (effective) { + case SINGLE_PAYLOADS: + case SINGLE_PAYLOAD: + case SINGLE_MEMO: + case MEMO_FIELDS_DATA_SUM: + case SINGLE_HEADER: + case SINGLE_SEARCH_ATTRIBUTES: + case WHOLE_MESSAGE: + sb.append(" if (msg.has").append(base).append("()) {\n"); + sb.append(" sink.check(\"") + .append(protoField) + .append("\", ") + .append(classToken) + .append(", ") + .append(singleSizeExpr(effective, "msg.get" + base + "()")) + .append(", ") + .append(ee) + .append(");\n"); + sb.append(" }\n"); + return; + case REPEATED_PAYLOAD: + case REPEATED_WHOLE_MESSAGE: + sb.append(" sink.check(\"") + .append(protoField) + .append("\", ") + .append(classToken) + .append(", PayloadLimitSizes.serializedSizeSum(msg.get") + .append(base) + .append("List()), ") + .append(ee) + .append(");\n"); + return; + case MAP_PAYLOAD: + sb.append(" sink.check(\"") + .append(protoField) + .append("\", ") + .append(classToken) + .append(", PayloadLimitSizes.mapPayloadDataSum(msg.get") + .append(base) + .append("Map()), ") + .append(ee) + .append(");\n"); + return; + case MAP_PAYLOADS: + sb.append(" sink.check(\"") + .append(protoField) + .append("\", ") + .append(classToken) + .append(", PayloadLimitSizes.mapPayloadsSum(msg.get") + .append(base) + .append("Map()), ") + .append(ee) + .append(");\n"); + return; + } + throw new AssertionError(effective); + } + + /** Size expression for an optional-singular leaf, given the getter expression for the value. */ + private static String singleSizeExpr(LeafKind kind, String getter) { + switch (kind) { + case SINGLE_PAYLOADS: + case SINGLE_PAYLOAD: + case SINGLE_MEMO: + case WHOLE_MESSAGE: + return "PayloadLimitSizes.serializedSize(" + getter + ")"; + case MEMO_FIELDS_DATA_SUM: + case SINGLE_HEADER: + return "PayloadLimitSizes.mapPayloadDataSum(" + getter + ".getFieldsMap())"; + case SINGLE_SEARCH_ATTRIBUTES: + return "PayloadLimitSizes.mapPayloadDataSum(" + getter + ".getIndexedFieldsMap())"; + case REPEATED_PAYLOAD: + case REPEATED_WHOLE_MESSAGE: + case MAP_PAYLOAD: + case MAP_PAYLOADS: + break; + } + throw new AssertionError(kind); + } + + private void emitStruct( + StringBuilder sb, FieldDescriptor f, StructShape shape, Descriptor child, int fi) { + String protoField = f.getName(); + String base = ProtoNames.base(f); + String childMethod = ProtoNames.methodName(child.getFullName()); + String childSrc = ProtoNames.sourceClassName(child); + switch (shape) { + case SINGLE: + sb.append(" if (msg.has").append(base).append("()) {\n"); + sb.append(" sink.enter(\"").append(protoField).append("\");\n"); + sb.append(" ") + .append(childMethod) + .append("(sink, msg.get") + .append(base) + .append("());\n"); + sb.append(" sink.exit();\n"); + sb.append(" }\n"); + return; + case REPEATED: + sb.append(" {\n"); + sb.append(" List<") + .append(childSrc) + .append("> __l") + .append(fi) + .append(" = msg.get") + .append(base) + .append("List();\n"); + sb.append(" for (int __i") + .append(fi) + .append(" = 0; __i") + .append(fi) + .append(" < __l") + .append(fi) + .append(".size(); __i") + .append(fi) + .append("++) {\n"); + sb.append(" sink.enter(\"") + .append(protoField) + .append("\", __i") + .append(fi) + .append(");\n"); + sb.append(" ") + .append(childMethod) + .append("(sink, __l") + .append(fi) + .append(".get(__i") + .append(fi) + .append("));\n"); + sb.append(" sink.exit();\n"); + sb.append(" }\n"); + sb.append(" }\n"); + return; + case MAP: + sb.append(" for (Map.Entry __e") + .append(fi) + .append(" : msg.get") + .append(base) + .append("Map().entrySet()) {\n"); + sb.append(" sink.enter(\"") + .append(protoField) + .append("\", __e") + .append(fi) + .append(".getKey());\n"); + sb.append(" ") + .append(childMethod) + .append("(sink, __e") + .append(fi) + .append(".getValue());\n"); + sb.append(" sink.exit();\n"); + sb.append(" }\n"); + return; + } + throw new AssertionError(shape); + } +} diff --git a/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoClosure.java b/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoClosure.java new file mode 100644 index 0000000000..038d53376f --- /dev/null +++ b/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoClosure.java @@ -0,0 +1,178 @@ +package io.temporal.internal.payload.limits.gen; + +import com.google.protobuf.Descriptors.Descriptor; +import com.google.protobuf.Descriptors.FieldDescriptor; +import com.google.protobuf.Descriptors.FileDescriptor; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * The message closure reachable from a set of seed services, and which of those messages can + * transitively contain a {@code Payload}. + * + *

Reachability is deliberately the same notion the payload visitor uses: a {@code + * google.protobuf.Any} counts as payload-bearing, because its contents are opaque here and may hold + * payloads. + * + *

A near-copy of this class exists in {@code temporal-sdk}, as {@code + * io.temporal.internal.payload.visitor.gen.ProtoClosure}, serving the payload visitor. See that + * class for why the duplication is deliberate. The copies differ only in how each obtains its + * {@code Descriptor}s (a descriptor set file here, compiled proto classes there) and in that this + * one computes reachability without the visitor generator's {@code FieldPlan}/{@code classify} + * model. Payload reachability must stay identical in both; verify any change by confirming + * both generators' output is byte-identical before and after it. + */ +final class ProtoClosure { + + private static final String PAYLOAD = "temporal.api.common.v1.Payload"; + private static final String PAYLOADS = "temporal.api.common.v1.Payloads"; + private static final String ANY = "google.protobuf.Any"; + + /** All non-map-entry messages in the closure, in discovery order. */ + final List allMessages; + + /** Full names of the messages that can transitively contain a payload. */ + private final Set reaches; + + private ProtoClosure(List allMessages, Set reaches) { + this.allMessages = allMessages; + this.reaches = reaches; + } + + /** Whether {@code d} can transitively contain a payload. */ + boolean reaches(Descriptor d) { + return reaches.contains(d.getFullName()); + } + + /** Builds the closure and payload-reachability set from the given seed file descriptors. */ + static ProtoClosure of(List seeds) { + List all = collectMessages(fileClosure(seeds)); + return new ProtoClosure(all, computeReachability(all)); + } + + // --- Descriptor discovery --- + + private static Set fileClosure(List seeds) { + Set seen = new LinkedHashSet<>(); + Deque queue = new ArrayDeque<>(seeds); + while (!queue.isEmpty()) { + FileDescriptor f = queue.poll(); + if (seen.add(f)) { + queue.addAll(f.getDependencies()); + } + } + return seen; + } + + private static List collectMessages(Set files) { + List result = new ArrayList<>(); + for (FileDescriptor f : files) { + for (Descriptor d : f.getMessageTypes()) { + collectMessages(d, result); + } + } + return result; + } + + private static void collectMessages(Descriptor d, List out) { + if (d.getOptions().getMapEntry()) { + return; // synthetic map entry type; handled via the owning map field + } + out.add(d); + for (Descriptor nested : d.getNestedTypes()) { + collectMessages(nested, out); + } + } + + // --- Reachability --- + + /** + * Least-fixpoint reachability over the message-reference graph. A message reaches a payload if it + * has a direct payload/Any field, or it references (via a message or map-message field) another + * message that does. Iterating to a fixpoint handles cycles (e.g. {@code Failure.cause}) + * correctly without over-approximating payload-free cycles. + */ + private static Set computeReachability(List all) { + Set reaches = new HashSet<>(); + Map> children = new HashMap<>(); + for (Descriptor d : all) { + boolean direct = false; + List refs = new ArrayList<>(); + for (FieldDescriptor f : d.getFields()) { + Descriptor referenced = referencedMessage(f); + if (referenced == null) { + if (carriesPayloadDirectly(f)) { + direct = true; + } + } else { + refs.add(referenced); + } + } + if (direct) { + reaches.add(d.getFullName()); + } + children.put(d.getFullName(), refs); + } + boolean changed = true; + while (changed) { + changed = false; + for (Descriptor d : all) { + if (reaches.contains(d.getFullName())) { + continue; + } + for (Descriptor c : children.get(d.getFullName())) { + if (reaches.contains(c.getFullName())) { + reaches.add(d.getFullName()); + changed = true; + break; + } + } + } + } + return reaches; + } + + /** Whether {@code f} holds payload data itself, in singular, repeated or map-valued form. */ + private static boolean carriesPayloadDirectly(FieldDescriptor f) { + String name = valueMessageName(f); + return PAYLOAD.equals(name) || PAYLOADS.equals(name) || ANY.equals(name); + } + + /** + * The Temporal message {@code f} refers to and should be recursed into, or {@code null} if it + * carries payload data directly or leads nowhere interesting. + */ + private static Descriptor referencedMessage(FieldDescriptor f) { + if (carriesPayloadDirectly(f)) { + return null; + } + Descriptor value = valueMessage(f); + if (value == null || !ProtoNames.isTemporal(value)) { + return null; + } + return value; + } + + /** The message type a field holds, unwrapping map values; {@code null} for non-message fields. */ + private static Descriptor valueMessage(FieldDescriptor f) { + if (f.isMapField()) { + FieldDescriptor value = f.getMessageType().findFieldByNumber(2); + return value.getJavaType() == FieldDescriptor.JavaType.MESSAGE + ? value.getMessageType() + : null; + } + return f.getJavaType() == FieldDescriptor.JavaType.MESSAGE ? f.getMessageType() : null; + } + + private static String valueMessageName(FieldDescriptor f) { + Descriptor value = valueMessage(f); + return value == null ? null : value.getFullName(); + } +} diff --git a/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoDescriptorSets.java b/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoDescriptorSets.java new file mode 100644 index 0000000000..cb32bf3d1f --- /dev/null +++ b/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoDescriptorSets.java @@ -0,0 +1,96 @@ +package io.temporal.internal.payload.limits.gen; + +import com.google.protobuf.DescriptorProtos.FileDescriptorProto; +import com.google.protobuf.DescriptorProtos.FileDescriptorSet; +import com.google.protobuf.Descriptors.DescriptorValidationException; +import com.google.protobuf.Descriptors.FileDescriptor; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Loads {@link FileDescriptor}s from a {@code protoc}-emitted descriptor set file. + * + *

The generator reads descriptors from a file rather than from compiled proto classes (via + * {@code SomeProto.getDescriptor()}) so that it does not depend on this module's own compiled + * output. The protos are generated into this module's main source set, so a generator that needed + * the compiled classes would form a dependency cycle with the very compilation that consumes its + * output. + */ +final class ProtoDescriptorSets { + private ProtoDescriptorSets() {} + + /** + * Loads every file in {@code descriptorSetFile}, keyed by proto file name (e.g. {@code + * temporal/api/workflowservice/v1/service.proto}). + * + *

The descriptor set must have been generated with imports included; a file referenced as a + * dependency but absent from the set is an error. + */ + static Map load(Path descriptorSetFile) + throws IOException, DescriptorValidationException { + FileDescriptorSet set; + try (InputStream in = Files.newInputStream(descriptorSetFile)) { + // Custom options are left as unknown fields; only standard options are read here. + set = FileDescriptorSet.parseFrom(in); + } + Map protos = new LinkedHashMap<>(); + for (FileDescriptorProto proto : set.getFileList()) { + // A file can appear more than once when several roots import it; the copies are identical. + protos.putIfAbsent(proto.getName(), proto); + } + Map built = new LinkedHashMap<>(); + for (String name : protos.keySet()) { + build(name, protos, built, new LinkedHashSet<>()); + } + return built; + } + + /** Builds {@code name} and, depth-first, the files it imports. */ + private static FileDescriptor build( + String name, + Map protos, + Map built, + Set building) + throws DescriptorValidationException { + FileDescriptor existing = built.get(name); + if (existing != null) { + return existing; + } + if (!building.add(name)) { + throw new IllegalStateException("cyclic proto import involving `" + name + "`"); + } + FileDescriptorProto proto = protos.get(name); + if (proto == null) { + throw new IllegalStateException( + "descriptor set is missing imported file `" + + name + + "`; it must be generated with descriptorSetOptions.includeImports = true"); + } + List dependencies = new ArrayList<>(); + for (String dependency : proto.getDependencyList()) { + dependencies.add(build(dependency, protos, built, building)); + } + FileDescriptor file = + FileDescriptor.buildFrom(proto, dependencies.toArray(new FileDescriptor[0])); + building.remove(name); + built.put(name, file); + return file; + } + + /** Looks up a file that must be present, with a message naming the missing file if it is not. */ + static FileDescriptor require(Map files, String name) { + FileDescriptor file = files.get(name); + if (file == null) { + throw new IllegalStateException("descriptor set does not contain `" + name + "`"); + } + return file; + } +} diff --git a/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoNames.java b/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoNames.java new file mode 100644 index 0000000000..2964eed986 --- /dev/null +++ b/temporal-serviceclient/src/payloadLimitsGenerator/java/io/temporal/internal/payload/limits/gen/ProtoNames.java @@ -0,0 +1,77 @@ +package io.temporal.internal.payload.limits.gen; + +import com.google.protobuf.Descriptors.Descriptor; +import com.google.protobuf.Descriptors.FieldDescriptor; +import java.util.ArrayDeque; +import java.util.Deque; + +/** + * Java naming rules for generated code, mirroring what {@code protoc} produces for a descriptor. + * + *

These duplicate the naming helpers in {@code temporal-sdk}'s {@code + * io.temporal.internal.payload.visitor.gen.PayloadVisitorGenerator}; see {@link ProtoClosure} for + * why the two generators are deliberately independent. Drift here is far less dangerous than drift + * in reachability: a wrong accessor name produces generated code that does not compile. + */ +final class ProtoNames { + private ProtoNames() {} + + /** Whether {@code d} is a Temporal-owned message (as opposed to a well-known/3rd-party type). */ + static boolean isTemporal(Descriptor d) { + return d.getFullName().startsWith("temporal."); + } + + /** Mirrors protoc's {@code UnderscoresToCamelCase}, used to derive Java accessor names. */ + static String camel(String input, boolean capNext) { + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < input.length(); i++) { + char c = input.charAt(i); + if (c >= 'a' && c <= 'z') { + sb.append(capNext ? Character.toUpperCase(c) : c); + capNext = false; + } else if (c >= 'A' && c <= 'Z') { + if (i == 0 && !capNext) { + sb.append(Character.toLowerCase(c)); + } else { + sb.append(c); + } + capNext = false; + } else if (c >= '0' && c <= '9') { + sb.append(c); + capNext = true; + } else { + capNext = true; + } + } + return sb.toString(); + } + + /** Capitalized accessor base, e.g. {@code schedule_activity} -> {@code ScheduleActivity}. */ + static String base(FieldDescriptor f) { + return camel(f.getName(), true); + } + + private static String javaPackage(Descriptor d) { + String pkg = d.getFile().getOptions().getJavaPackage(); + if (pkg == null || pkg.isEmpty()) { + throw new IllegalStateException("message " + d.getFullName() + " has no java_package option"); + } + return pkg; + } + + /** + * Source-form class name, e.g. {@code io.temporal.api.common.v1.Payload.ExternalPayloadDetails}. + */ + static String sourceClassName(Descriptor d) { + Deque names = new ArrayDeque<>(); + for (Descriptor c = d; c != null; c = c.getContainingType()) { + names.addFirst(c.getName()); + } + return javaPackage(d) + "." + String.join(".", names); + } + + /** Name of the generated per-message method for {@code full}, a descriptor full name. */ + static String methodName(String full) { + return "visit_" + full.replace('.', '_'); + } +} diff --git a/temporal-serviceclient/src/test/java/io/temporal/internal/payload/limits/PayloadLimitValidatorTest.java b/temporal-serviceclient/src/test/java/io/temporal/internal/payload/limits/PayloadLimitValidatorTest.java new file mode 100644 index 0000000000..691c610961 --- /dev/null +++ b/temporal-serviceclient/src/test/java/io/temporal/internal/payload/limits/PayloadLimitValidatorTest.java @@ -0,0 +1,458 @@ +package io.temporal.internal.payload.limits; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import com.google.protobuf.Any; +import com.google.protobuf.ByteString; +import io.temporal.api.command.v1.Command; +import io.temporal.api.command.v1.CompleteWorkflowExecutionCommandAttributes; +import io.temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes; +import io.temporal.api.command.v1.FailWorkflowExecutionCommandAttributes; +import io.temporal.api.command.v1.ModifyWorkflowPropertiesCommandAttributes; +import io.temporal.api.command.v1.RecordMarkerCommandAttributes; +import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributes; +import io.temporal.api.command.v1.ScheduleNexusOperationCommandAttributes; +import io.temporal.api.common.v1.Header; +import io.temporal.api.common.v1.Memo; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.common.v1.SearchAttributes; +import io.temporal.api.failure.v1.Failure; +import io.temporal.api.protocol.v1.Message; +import io.temporal.api.query.v1.WorkflowQueryResult; +import io.temporal.api.sdk.v1.UserMetadata; +import io.temporal.api.update.v1.Input; +import io.temporal.api.update.v1.Request; +import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.RespondActivityTaskFailedRequest; +import io.temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedRequest; +import io.temporal.api.workflowservice.v1.StartWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.TerminateWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.UpdateWorkflowExecutionRequest; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.junit.Test; + +public class PayloadLimitValidatorTest { + + // --- helpers --------------------------------------------------------------- + + private static Payload payload(int dataLen) { + return Payload.newBuilder().setData(ByteString.copyFrom(new byte[dataLen])).build(); + } + + private static Payloads payloads(int dataLen) { + return Payloads.newBuilder().addPayloads(payload(dataLen)).build(); + } + + private static Memo memoWith(String key, int dataLen) { + return Memo.newBuilder().putFields(key, payload(dataLen)).build(); + } + + /** blobWarn = memoWarn = 10, with the given error thresholds. */ + private static PayloadLimits workerLimits(long blobError, long memoError) { + return new PayloadLimits(10, blobError, 10, memoError); + } + + private static RespondWorkflowTaskCompletedRequest wftWithCommand(Command command) { + return RespondWorkflowTaskCompletedRequest.newBuilder().addCommands(command).build(); + } + + // --- end-to-end validation via PayloadLimitValidator ----------------------- + + @Test + public void blobFieldOverErrorLimitIsReported() { + StartWorkflowExecutionRequest req = + StartWorkflowExecutionRequest.newBuilder().setInput(payloads(1000)).build(); + Optional v = PayloadLimitValidator.validate(req, workerLimits(100, 100)); + assertTrue(v.isPresent()); + assertEquals(LimitClass.BLOB, v.get().getLimitClass()); + assertEquals(LimitSeverity.ERROR, v.get().getSeverity()); + assertEquals("input", v.get().getPath()); + assertTrue(v.get().getSize() > 100); + } + + @Test + public void memoFieldUsesMemoLimit() { + // A memo over the memo error limit but well under the (huge) blob error limit must still error, + // proving the memo class routes to the memo threshold. + StartWorkflowExecutionRequest req = + StartWorkflowExecutionRequest.newBuilder().setMemo(memoWith("k", 50)).build(); + PayloadLimits limits = new PayloadLimits(10, 1_000_000, 10, 20); + Optional v = PayloadLimitValidator.validate(req, limits); + assertTrue(v.isPresent()); + assertEquals(LimitClass.MEMO, v.get().getLimitClass()); + assertEquals("memo", v.get().getPath()); + } + + @Test + public void warnOnlyClassifiedFieldNeverErrors() { + // RespondActivityTaskFailed.failure is classified warn-only, so even with an error limit a huge + // failure produces no error-level violation. + RespondActivityTaskFailedRequest req = + RespondActivityTaskFailedRequest.newBuilder() + .setFailure(Failure.newBuilder().setMessage(repeat("x", 10_000))) + .build(); + assertFalse(PayloadLimitValidator.validate(req, workerLimits(100, 100)).isPresent()); + } + + @Test + public void underLimitIsOk() { + StartWorkflowExecutionRequest req = + StartWorkflowExecutionRequest.newBuilder().setInput(payloads(5)).build(); + assertFalse(PayloadLimitValidator.validate(req, workerLimits(100_000, 100_000)).isPresent()); + } + + @Test + public void requestWithoutPayloadFieldsIsIgnored() { + assertFalse( + PayloadLimitValidator.validate( + DescribeWorkflowExecutionRequest.getDefaultInstance(), workerLimits(1, 1)) + .isPresent()); + } + + @Test + public void blobClassedMemoIsMeasuredAsFieldsDataSum() { + // ModifyWorkflowProperties.upserted_memo is blob-classed, so it is measured as the data-sum of + // its fields (key bytes + payload data bytes), NOT the whole-Memo proto size. + Memo memo = + Memo.newBuilder().putFields("ab", payload(10)).putFields("cde", payload(20)).build(); + RespondWorkflowTaskCompletedRequest req = + wftWithCommand( + Command.newBuilder() + .setModifyWorkflowPropertiesCommandAttributes( + ModifyWorkflowPropertiesCommandAttributes.newBuilder().setUpsertedMemo(memo)) + .build()); + // data-sum = (2 + 10) + (3 + 20) = 35; blobError = 30 -> error, classified BLOB. + Optional v = + PayloadLimitValidator.validate(req, workerLimits(30, 1_000_000)); + assertTrue(v.isPresent()); + assertEquals(LimitClass.BLOB, v.get().getLimitClass()); + assertEquals( + "commands[0].modify_workflow_properties_command_attributes.upserted_memo", + v.get().getPath()); + assertEquals(35, v.get().getSize()); + } + + @Test + public void markerDetailsMapIsMeasuredAsPayloadsSum() { + RespondWorkflowTaskCompletedRequest req = + wftWithCommand( + Command.newBuilder() + .setRecordMarkerCommandAttributes( + RecordMarkerCommandAttributes.newBuilder().putDetails("marker", payloads(1000))) + .build()); + Optional v = PayloadLimitValidator.validate(req, workerLimits(100, 100)); + assertTrue(v.isPresent()); + assertEquals(LimitClass.BLOB, v.get().getLimitClass()); + assertEquals("commands[0].record_marker_command_attributes.details", v.get().getPath()); + } + + @Test + public void singlePayloadFieldIsMeasuredAsPayloadSize() { + // ScheduleNexusOperation.input is a single Payload (not Payloads). + RespondWorkflowTaskCompletedRequest req = + wftWithCommand( + Command.newBuilder() + .setScheduleNexusOperationCommandAttributes( + ScheduleNexusOperationCommandAttributes.newBuilder().setInput(payload(1000))) + .build()); + Optional v = PayloadLimitValidator.validate(req, workerLimits(100, 100)); + assertTrue(v.isPresent()); + assertEquals(LimitClass.BLOB, v.get().getLimitClass()); + assertEquals( + "commands[0].schedule_nexus_operation_command_attributes.input", v.get().getPath()); + } + + @Test + public void wholeFailureIsMeasuredAsMessageSize() { + RespondWorkflowTaskCompletedRequest req = + wftWithCommand( + Command.newBuilder() + .setFailWorkflowExecutionCommandAttributes( + FailWorkflowExecutionCommandAttributes.newBuilder() + .setFailure(Failure.newBuilder().setMessage(repeat("x", 1000)))) + .build()); + Optional v = PayloadLimitValidator.validate(req, workerLimits(100, 100)); + assertTrue(v.isPresent()); + assertEquals(LimitClass.BLOB, v.get().getLimitClass()); + assertEquals( + "commands[0].fail_workflow_execution_command_attributes.failure", v.get().getPath()); + } + + @Test + public void tmprl1103MessageText() { + StartWorkflowExecutionRequest req = + StartWorkflowExecutionRequest.newBuilder().setInput(payloads(1000)).build(); + PayloadLimitViolation v = PayloadLimitValidator.validate(req, workerLimits(100, 100)).get(); + assertEquals( + "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit.", + v.getMessage()); + } + + // --- CollectingSink: classification, independent of logging/early-return ---- + + @Test + public void collectingSinkClassifiesErrorVsWarning() { + CollectingSink sink = new CollectingSink(workerLimits(100, 100)); // blobWarn=10, blobError=100 + sink.check("over_error", LimitClass.BLOB, 200, true); + sink.check("over_warn", LimitClass.BLOB, 50, true); + sink.check("under_warn", LimitClass.BLOB, 5, true); + assertEquals(1, sink.getErrors().size()); + assertEquals("over_error", sink.getErrors().get(0).getPath()); + assertEquals(100, sink.getErrors().get(0).getLimit()); + assertEquals(1, sink.getWarnings().size()); + assertEquals("over_warn", sink.getWarnings().get(0).getPath()); + assertEquals(10, sink.getWarnings().get(0).getLimit()); + } + + @Test + public void collectingSinkWarnOnlyFieldNeverErrors() { + CollectingSink sink = new CollectingSink(workerLimits(100, 100)); + sink.check("warn_only", LimitClass.BLOB, 5000, false); // enforceError = false + assertTrue(sink.getErrors().isEmpty()); + assertEquals(1, sink.getWarnings().size()); + } + + @Test + public void collectingSinkNoErrorLimitOnlyWarns() { + CollectingSink sink = new CollectingSink(new PayloadLimits(100, 0, 0, 0)); + sink.check("big", LimitClass.BLOB, 101, true); // error threshold 0 disables errors + assertTrue(sink.getErrors().isEmpty()); + assertEquals(1, sink.getWarnings().size()); + } + + @Test + public void collectingSinkZeroWarnDisablesWarnings() { + CollectingSink sink = new CollectingSink(PayloadLimits.none()); + sink.check("big", LimitClass.BLOB, 5000, true); + assertTrue(sink.getErrors().isEmpty()); + assertTrue(sink.getWarnings().isEmpty()); + } + + @Test + public void collectingSinkRoutesMemoToMemoLimit() { + CollectingSink sink = new CollectingSink(workerLimits(1_000_000, 20)); + sink.check("blob_field", LimitClass.BLOB, 100, true); // fine: huge blob limit + sink.check("memo_field", LimitClass.MEMO, 100, true); // errors: tiny memo limit + assertEquals(1, sink.getErrors().size()); + assertEquals(LimitClass.MEMO, sink.getErrors().get(0).getLimitClass()); + assertEquals("memo_field", sink.getErrors().get(0).getPath()); + } + + // --- Which fields get visited (order-independent) -------------------------- + + @Test + public void visitsPayloadFieldsOfEachCommandWithPaths() { + RespondWorkflowTaskCompletedRequest req = + RespondWorkflowTaskCompletedRequest.newBuilder() + .addCommands( + Command.newBuilder() + .setScheduleActivityTaskCommandAttributes( + ScheduleActivityTaskCommandAttributes.newBuilder().setInput(payloads(1)))) + .addCommands( + Command.newBuilder() + .setCompleteWorkflowExecutionCommandAttributes( + CompleteWorkflowExecutionCommandAttributes.newBuilder() + .setResult(payloads(1)))) + .build(); + RecordingSink sink = new RecordingSink(); + GeneratedPayloadLimitValidator.dispatch(sink, req); + assertEquals( + Arrays.asList( + "commands[0].schedule_activity_task_command_attributes.input", + "commands[1].complete_workflow_execution_command_attributes.result"), + sink.sorted()); + } + + @Test + public void visitsOnlyPresentFields() { + // Only input is set; memo/search_attributes/etc. are absent and must not be visited. + StartWorkflowExecutionRequest req = + StartWorkflowExecutionRequest.newBuilder().setInput(payloads(1)).build(); + RecordingSink sink = new RecordingSink(); + GeneratedPayloadLimitValidator.dispatch(sink, req); + assertEquals(Collections.singletonList("input"), sink.sorted()); + } + + @Test + public void visitsProtocolMessageBody() { + // Message.body is a google.protobuf.Any reached because Message is a forced whole-message leaf + // and the parent recurses into `messages`. + RespondWorkflowTaskCompletedRequest req = + RespondWorkflowTaskCompletedRequest.newBuilder() + .addMessages(Message.newBuilder().setBody(Any.getDefaultInstance())) + .build(); + RecordingSink sink = new RecordingSink(); + GeneratedPayloadLimitValidator.dispatch(sink, req); + assertEquals(Collections.singletonList("messages[0].body"), sink.sorted()); + } + + // --- NOT_VALIDATED fields must never be checked --------------------------- + // + // These are the tests that catch an accidental reclassification: a field the server does not + // size-check (or checks in a way the SDK cannot replicate) must produce no callback at all, so + // asserting the exact visited set is stronger than asserting no violation. + + @Test + public void startWorkflowVisitsOnlyItsValidatedFields() { + StartWorkflowExecutionRequest req = + StartWorkflowExecutionRequest.newBuilder() + .setInput(payloads(1)) // blob + .setMemo(memoWith("k", 1)) // memo + .setHeader(Header.newBuilder().putFields("h", payload(1))) // metric only + .setSearchAttributes( + SearchAttributes.newBuilder().putIndexedFields("sa", payload(1))) // server-state + .setLastCompletionResult(payloads(1)) // carry-over, not checked here + .setContinuedFailure(Failure.newBuilder().setMessage("boom")) // carry-over + .setUserMetadata( + UserMetadata.newBuilder() + .setSummary(payload(1)) + .setDetails(payload(1))) // dedicated non-fetchable limits + .build(); + RecordingSink sink = new RecordingSink(); + GeneratedPayloadLimitValidator.dispatch(sink, req); + assertEquals(Arrays.asList("input", "memo"), sink.sorted()); + } + + @Test + public void continueAsNewVisitsOnlyItsValidatedFields() { + RespondWorkflowTaskCompletedRequest req = + wftWithCommand( + Command.newBuilder() + .setContinueAsNewWorkflowExecutionCommandAttributes( + ContinueAsNewWorkflowExecutionCommandAttributes.newBuilder() + .setInput(payloads(1)) // blob + .setMemo(memoWith("k", 1)) // memo + .setHeader(Header.newBuilder().putFields("h", payload(1))) + .setSearchAttributes( + SearchAttributes.newBuilder().putIndexedFields("sa", payload(1))) + .setFailure(Failure.newBuilder().setMessage("boom")) + .setLastCompletionResult(payloads(1))) + .build()); + RecordingSink sink = new RecordingSink(); + GeneratedPayloadLimitValidator.dispatch(sink, req); + assertEquals( + Arrays.asList( + "commands[0].continue_as_new_workflow_execution_command_attributes.input", + "commands[0].continue_as_new_workflow_execution_command_attributes.memo"), + sink.sorted()); + } + + @Test + public void recordMarkerVisitsOnlyDetails() { + RespondWorkflowTaskCompletedRequest req = + wftWithCommand( + Command.newBuilder() + .setRecordMarkerCommandAttributes( + RecordMarkerCommandAttributes.newBuilder() + .putDetails("marker", payloads(1)) // blob + .setHeader(Header.newBuilder().putFields("h", payload(1))) + .setFailure(Failure.newBuilder().setMessage("boom"))) + .build()); + RecordingSink sink = new RecordingSink(); + GeneratedPayloadLimitValidator.dispatch(sink, req); + assertEquals( + Collections.singletonList("commands[0].record_marker_command_attributes.details"), + sink.sorted()); + } + + @Test + public void requestsWithOnlyNotValidatedFieldsProduceNoChecks() { + RecordingSink terminate = new RecordingSink(); + GeneratedPayloadLimitValidator.dispatch( + terminate, + TerminateWorkflowExecutionRequest.newBuilder().setDetails(payloads(5000)).build()); + assertEquals(Collections.emptyList(), terminate.sorted()); + + // Update args are recorded as a metric by the frontend; the size the server enforces is the + // protocol Message body on delivery, which is checked on the worker's completion instead. + RecordingSink update = new RecordingSink(); + GeneratedPayloadLimitValidator.dispatch( + update, + UpdateWorkflowExecutionRequest.newBuilder() + .setRequest( + Request.newBuilder() + .setInput( + Input.newBuilder() + .setArgs(payloads(5000)) + .setHeader(Header.newBuilder().putFields("h", payload(1))))) + .build()); + assertEquals(Collections.emptyList(), update.sorted()); + } + + @Test + public void mapKeyedFieldsRenderTheKeyInThePath() { + RespondWorkflowTaskCompletedRequest req = + RespondWorkflowTaskCompletedRequest.newBuilder() + .putQueryResults( + "query-id", WorkflowQueryResult.newBuilder().setAnswer(payloads(1)).build()) + .build(); + RecordingSink sink = new RecordingSink(); + GeneratedPayloadLimitValidator.dispatch(sink, req); + assertEquals(Collections.singletonList("query_results[query-id].answer"), sink.sorted()); + } + + // --- size helpers --------------------------------------------------------- + + @Test + public void mapPayloadDataSumCountsUtf8KeyBytesAndRawData() { + // The server sums len(key) + len(payload.data) over Go strings, i.e. UTF-8 bytes: the 2-char + // key below is 6 bytes, so a char count would under-measure it. + Map fields = new HashMap<>(); + fields.put("\u00e9\u4e2d", payload(10)); // 2 + 3 UTF-8 bytes + assertEquals(15, PayloadLimitSizes.mapPayloadDataSum(fields)); + } + + private static String repeat(String s, int n) { + StringBuilder sb = new StringBuilder(s.length() * n); + for (int i = 0; i < n; i++) { + sb.append(s); + } + return sb.toString(); + } + + /** A sink that records the path of each visited field (order is not significant). */ + private static final class RecordingSink implements PayloadLimitSink { + private final PayloadPath path = new PayloadPath(); + private final List visited = new ArrayList<>(); + + @Override + public void check(String fieldName, LimitClass limitClass, long size, boolean enforceError) { + visited.add(path.leaf(fieldName)); + } + + @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 sorted() { + List v = new ArrayList<>(visited); + Collections.sort(v); + return v; + } + } +}