fix: bound telemetry truncation for built-in exporters

telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetry.java

@@ -17,6 +17,7 @@
 import software.amazon.lambda.durable.plugin.InvocationInfo;
 import software.amazon.lambda.durable.plugin.InvocationStatus;
 import software.amazon.lambda.durable.plugin.OperationChangeItemInfo;
+import software.amazon.lambda.durable.telemetry.exporters.CloudWatchLogsExporter;
 import software.amazon.lambda.durable.telemetry.exporters.LambdaLogExporter;
 import software.amazon.lambda.durable.telemetry.internal.FatalErrors;
 
@@ -80,17 +81,32 @@ static void exportRecord(DurableExecutionTelemetryRecord record, TelemetryExport
             // so without this a custom exporter that mutates operations or nested content would corrupt every
             // other exporter's view of the same record.
             DurableExecutionTelemetryRecord isolated = record.deepCopy();
-            DurableExecutionTelemetryRecord shaped = exporter instanceof LambdaLogExporter lambdaLog
-                    ? Truncation.truncateMeasured(isolated, exporter.maxRecordSizeBytes(), lambdaLog::renderedSize)
-                    : Truncation.truncateRecord(isolated, exporter.maxRecordSizeBytes(), exporter::render);
-            exporter.export(shaped);
+            exporter.export(shapedForExport(isolated, exporter));
         } catch (Throwable t) {
             // Ordinary Errors (such as missing optional dependencies) are isolated too.
             // logSafely rethrows VM/thread termination before logging or continuing.
             logSafely("exporter failed", t);
         }
     }
 
+    /**
+     * Built-in exporters render the record JSON, or that JSON embedded once as a string. The shrink bound covers two
+     * copies so {@code OperationsFormat.BOTH} (the array and the by-name map) drops the same operations as checking
+     * every step. {@link LambdaLogExporter} already measures its line with one copy. A custom exporter is not in the
+     * built-in package and keeps a check after every drop.
+     */
+    static DurableExecutionTelemetryRecord shapedForExport(
+            DurableExecutionTelemetryRecord record, TelemetryExporter exporter) {
+        Integer limit = exporter.maxRecordSizeBytes();
+        if (exporter instanceof LambdaLogExporter lambdaLog) {
+            return Truncation.truncateMeasured(record, limit, lambdaLog::renderedSize);
+        }
+        if (exporter.getClass().getPackageName().equals(CloudWatchLogsExporter.class.getPackageName())) {
+            return Truncation.truncateMeasured(record, limit, r -> Json.byteSize(exporter.render(r)), 2);
+        }
+        return Truncation.truncateRecord(record, limit, exporter::render);
+    }
+
     /**
      * Applies a user-supplied result transform to an operation's checkpointed (serialized JSON) result. Parses the JSON
      * before handing it to the transform, so the transform always receives a <em>detached, JSON-compatible</em> value

telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/Truncation.java

@@ -36,7 +36,7 @@ public static DurableExecutionTelemetryRecord truncateRecord(
             DurableExecutionTelemetryRecord record,
             Integer maxBytes,
             Function<DurableExecutionTelemetryRecord, Object> render) {
-        return truncate(record, maxBytes, r -> Json.byteSize(render.apply(r)), false);
+        return truncate(record, maxBytes, r -> Json.byteSize(render.apply(r)), false, 1);
     }
 
     /**
@@ -50,7 +50,21 @@ static DurableExecutionTelemetryRecord truncateMeasured(
             DurableExecutionTelemetryRecord record,
             Integer maxBytes,
             Function<DurableExecutionTelemetryRecord, Integer> size) {
-        return truncate(record, maxBytes, size, true);
+        return truncateMeasured(record, maxBytes, size, 1);
+    }
+
+    /**
+     * {@link #truncateMeasured(DurableExecutionTelemetryRecord, Integer, Function)} when the rendering contains
+     * {@code representations} copies of each operation, such as {@code OperationsFormat.BOTH} (the operations array and
+     * the by-name map). The shrink bound is that many times one copy, so a check is skipped only when it still could
+     * not succeed. A larger bound skips fewer checks and drops the same operations.
+     */
+    static DurableExecutionTelemetryRecord truncateMeasured(
+            DurableExecutionTelemetryRecord record,
+            Integer maxBytes,
+            Function<DurableExecutionTelemetryRecord, Integer> size,
+            int representations) {
+        return truncate(record, maxBytes, size, true, representations);
     }
 
     /**
@@ -64,7 +78,8 @@ private static DurableExecutionTelemetryRecord truncate(
             DurableExecutionTelemetryRecord record,
             Integer maxBytes,
             Function<DurableExecutionTelemetryRecord, Integer> size,
-            boolean bounded) {
+            boolean bounded,
+            int representations) {
         if (maxBytes == null || maxBytes <= 0) {
             return record;
         }
@@ -94,7 +109,7 @@ record = base;
         boolean[] kept = new boolean[ops.size()];
         java.util.Arrays.fill(kept, true);
         int[] order = oldestFirstOrder(ops);
-        long[] shrinkBound = bounded ? shrinkBounds(ops) : null;
+        long[] shrinkBound = bounded ? shrinkBounds(ops, representations) : null;
         Meter meter = new Meter(size, maxBytes, bounded);
 
         boolean[] anyResultDropped = {false};
@@ -159,23 +174,23 @@ private static ErrorInfo withoutCause(ErrorInfo error) {
 
     /**
      * Upper bound, per operation, on how much dropping its result or the whole operation can shrink the rendering: the
-     * operation's JSON and its name, sized as if escaped again inside a JSON string, plus {@link #SHRINK_SLACK_BYTES}.
-     * Each operation is serialized once.
+     * operation's JSON and its name, sized as if escaped again inside a JSON string, plus {@link #SHRINK_SLACK_BYTES},
+     * times {@code representations} when the rendering contains that many copies. Each operation is serialized once.
      */
-    private static long[] shrinkBounds(List<OperationRecord> ops) {
+    private static long[] shrinkBounds(List<OperationRecord> ops, int representations) {
         long[] bounds = new long[ops.size()];
         for (int i = 0; i < bounds.length; i++) {
             OperationRecord op = ops.get(i);
-            long bound = SHRINK_SLACK_BYTES;
+            long one = SHRINK_SLACK_BYTES;
             try {
-                bound += Json.escapedUtf8Size(Json.stringify(op.toWireMap()));
+                one += Json.escapedUtf8Size(Json.stringify(op.toWireMap()));
                 if (op.name() != null) {
-                    bound += Json.escapedUtf8Size(Json.stringify(op.name()));
+                    one += Json.escapedUtf8Size(Json.stringify(op.name()));
                 }
+                bounds[i] = one * representations;
             } catch (IllegalStateException e) {
-                bound = Long.MAX_VALUE / 4; // Unknown: never skip a check after dropping this operation.
+                bounds[i] = Long.MAX_VALUE / 4; // Unknown: never skip a check after dropping this operation.
             }
-            bounds[i] = bound;
         }
         return bounds;
     }

telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/TruncationCostTest.java

@@ -0,0 +1,162 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.telemetry;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Test;
+import software.amazon.lambda.durable.telemetry.exporters.CloudWatchLogsExporter;
+import software.amazon.lambda.durable.telemetry.exporters.OTelExporter;
+import software.amazon.lambda.durable.telemetry.exporters.OperationsFormat;
+import software.amazon.lambda.durable.telemetry.exporters.RenderCountingExporter;
+
+/**
+ * Size-limited built-in exporters other than {@code LambdaLogExporter} used to measure the whole record once per
+ * operation. Issue #877: 4,000 simple steps at the CloudWatch limit took about 15 s.
+ */
+class TruncationCostTest {
+
+    private static final int MANY_STEPS = 2_000;
+    private static final int CLOUDWATCH_LIMIT = 256_000;
+    /** A bounded pass checks a few dozen times. One check per operation is thousands. */
+    private static final int FEW_MEASUREMENTS = 100;
+
+    @Test
+    void cloudWatchByNameShapeMeasuresAFewTimesAndKeepsTheExactOperations() {
+        assertEquals(RenderCountingExporter.class.getPackageName(), CloudWatchLogsExporter.class.getPackageName());
+        assertEquals(RenderCountingExporter.class.getPackageName(), OTelExporter.class.getPackageName());
+
+        RenderCountingExporter exporter = new RenderCountingExporter(OperationsFormat.BY_NAME, CLOUDWATCH_LIMIT);
+        DurableExecutionTelemetryRecord record = steps(MANY_STEPS, false);
+        assertSameKeptOperations(record, exporter, FEW_MEASUREMENTS);
+    }
+
+    /**
+     * {@code OperationsFormat.BOTH} and the OTel body (that JSON embedded as one string) must drop the same operations
+     * as measuring every step. A bound that covers only one representation drops extra operations when each one carries
+     * a large error.
+     */
+    @Test
+    void bothFormatAndOtelBodyMatchTheExactPass() {
+        DurableExecutionTelemetryRecord record = steps(30, true);
+        assertSameKeptOperations(record, new RenderCountingExporter(OperationsFormat.BOTH, 80_000), FEW_MEASUREMENTS);
+
+        OTelExporter otel = OTelExporter.builder()
+                .endpoint("http://127.0.0.1:9/v1/logs")
+                .operationsFormat(OperationsFormat.BOTH)
+                .build();
+        assertSameKeptOperations(record, new RenderCountingExporter(otel::render, 80_000), FEW_MEASUREMENTS);
+    }
+
+    /** A custom rendering stays on the exact pass: one check per step, same operations. */
+    @Test
+    void customRenderKeepsMeasuringEveryStep() {
+        DurableExecutionTelemetryRecord record = steps(80, false);
+        CustomExporter exporter = new CustomExporter(12_000);
+        DurableExecutionTelemetry.exportRecord(record.deepCopy(), exporter);
+        assertNotNull(exporter.exported, "export did not run");
+        int measurements = exporter.renders.get();
+        assertEquals(names(truncateExactly(record, exporter)), names(exporter.exported));
+        assertTrue(
+                measurements > record.operations().size(),
+                measurements + " measurements; the exact pass checks every operation");
+    }
+
+    private static void assertSameKeptOperations(
+            DurableExecutionTelemetryRecord record, RenderCountingExporter exporter, int maxMeasurements) {
+        DurableExecutionTelemetry.exportRecord(record.deepCopy(), exporter);
+        assertNotNull(exporter.exported, "export did not run");
+        int measurements = exporter.renders.get();
+        DurableExecutionTelemetryRecord exact = truncateExactly(record, exporter);
+        assertEquals(exact.droppedOperations, exporter.exported.droppedOperations);
+        assertEquals(names(exact), names(exporter.exported));
+        assertNotNull(exporter.exported.droppedOperations, "fixture must be over the limit");
+        assertTrue(
+                measurements <= maxMeasurements,
+                measurements + " measurements for " + record.operations().size() + " operations");
+    }
+
+    private static DurableExecutionTelemetryRecord truncateExactly(
+            DurableExecutionTelemetryRecord record, TelemetryExporter exporter) {
+        return Truncation.truncateRecord(record.deepCopy(), exporter.maxRecordSizeBytes(), exporter::render);
+    }
+
+    private static List<String> names(DurableExecutionTelemetryRecord record) {
+        List<String> names = new ArrayList<>();
+        for (OperationRecord op : record.operations()) {
+            names.add(op.name());
+        }
+        return names;
+    }
+
+    /** {@code count} succeeded steps. When {@code withErrors} is set, each step has its own large error. */
+    private static DurableExecutionTelemetryRecord steps(int count, boolean withErrors) {
+        DurableExecutionTelemetryRecord record = new DurableExecutionTelemetryRecord();
+        record.executionArn =
+                "arn:aws:lambda:us-east-1:123456789012:function:demo:$LATEST/durable-execution/run-1/inv-1";
+        record.emittedAt = "2026-10-28T00:00:00.000Z";
+        record.executionName = "run-1";
+        record.functionName = "demo";
+        record.functionQualifier = "$LATEST";
+        record.region = "us-east-1";
+        record.accountId = "123456789012";
+        record.durationMs = 1_000L;
+        record.status = "SUCCEEDED";
+        Instant start = Instant.parse("2026-10-28T00:00:00Z");
+        for (int i = 0; i < count; i++) {
+            Instant at = start.plusMillis(i * 10L);
+            OperationRecord op = new OperationRecord()
+                    .id("op-" + i)
+                    .name("step-" + i)
+                    .type("STEP")
+                    .subType("Step")
+                    .status("SUCCEEDED")
+                    .startTime(at.toString())
+                    .endTime(at.plusMillis(3).toString())
+                    .durationMs(3L)
+                    .attempt(1);
+            if (withErrors) {
+                op.status("FAILED").error(new ErrorInfo("E" + i, "e".repeat(3_000)));
+            }
+            record.addOperation(op);
+        }
+        return record;
+    }
+
+    /** Repeats the by-name map so one representation's shrink bound does not cover a drop. */
+    private static final class CustomExporter implements TelemetryExporter {
+        final AtomicInteger renders = new AtomicInteger();
+        DurableExecutionTelemetryRecord exported;
+        private final int maxBytes;
+
+        CustomExporter(int maxBytes) {
+            this.maxBytes = maxBytes;
+        }
+
+        @Override
+        public Integer maxRecordSizeBytes() {
+            return maxBytes;
+        }
+
+        @Override
+        public Object render(DurableExecutionTelemetryRecord record) {
+            renders.incrementAndGet();
+            List<Object> copies = new ArrayList<>();
+            for (int i = 0; i < 8; i++) {
+                copies.add(record.toByNameWireMap());
+            }
+            return copies;
+        }
+
+        @Override
+        public void export(DurableExecutionTelemetryRecord record) {
+            exported = record;
+        }
+    }
+}

telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/exporters/RenderCountingExporter.java

@@ -0,0 +1,45 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.telemetry.exporters;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetryRecord;
+import software.amazon.lambda.durable.telemetry.TelemetryExporter;
+
+/**
+ * Test exporter in this package so the export path treats it as a built-in. Counts {@link #render} calls, which is how
+ * truncation measures a record.
+ */
+public final class RenderCountingExporter implements TelemetryExporter {
+    public final AtomicInteger renders = new AtomicInteger();
+    public DurableExecutionTelemetryRecord exported;
+
+    private final Function<DurableExecutionTelemetryRecord, Object> renderFn;
+    private final int maxBytes;
+
+    public RenderCountingExporter(OperationsFormat format, int maxBytes) {
+        this(format::apply, maxBytes);
+    }
+
+    public RenderCountingExporter(Function<DurableExecutionTelemetryRecord, Object> renderFn, int maxBytes) {
+        this.renderFn = renderFn;
+        this.maxBytes = maxBytes;
+    }
+
+    @Override
+    public Integer maxRecordSizeBytes() {
+        return maxBytes;
+    }
+
+    @Override
+    public Object render(DurableExecutionTelemetryRecord record) {
+        renders.incrementAndGet();
+        return renderFn.apply(record);
+    }
+
+    @Override
+    public void export(DurableExecutionTelemetryRecord record) {
+        exported = record;
+    }
+}