fix(telemetry): write fixed-width UTC timestamps

telemetry-exporter/README.md

@@ -65,8 +65,8 @@ Writes one line per record to `System.out`, in the Lambda JSON log envelope:
 {"timestamp":"2026-10-21T08:00:00.000Z","level":"INFO","message":"<compact record JSON>"}
 ```
 
-`timestamp` is UTC ISO-8601 with exactly three fraction digits. The keys appear in that order. The
-[Lambda Java logging docs](https://docs.aws.amazon.com/lambda/latest/dg/java-logging.html) and the
+`timestamp` has the same format as the record timestamps (see [Timestamps](#timestamps)). The keys appear in that
+order. The [Lambda Java logging docs](https://docs.aws.amazon.com/lambda/latest/dg/java-logging.html) and the
 [Lambda log format docs](https://docs.aws.amazon.com/lambda/latest/dg/monitoring-cloudwatchlogs-logformat.html)
 list `LambdaLogger` and Log4j2 as the Java methods that Lambda captures as structured JSON, so `System.out` is not
 wrapped and the exporter writes the envelope itself. Lambda does not double-encode logs that are already JSON, so the
@@ -202,6 +202,23 @@ FileExporter.builder().directory("/mnt/efs/durable-execution-telemetry").mode(Fi
 `S3Exporter` writes `{prefix}{partition}{executionName}.json` (IAM: `s3:PutObject`). `CloudWatchLogsExporter` writes
 one event per record to `{logStreamPrefix}YYYY/MM/DD` (IAM: `logs:CreateLogStream`, `logs:PutLogEvents`).
 
+## Timestamps
+
+Every timestamp in a record has one format: `YYYY-MM-DDTHH:MM:SS.sssZ`, for example `2026-10-23T18:42:44.078Z`.
+This covers `emittedAt`, `startTime` and `endTime` on the record, and `startTime` and `endTime` on each operation.
+
+- The time is in UTC, and the value always ends with the literal `Z`.
+- The fraction always has exactly three digits. A whole second is written as `.000`.
+- Digits below the millisecond are dropped, not rounded. So `.078757096` is written as `.078`.
+
+This is the format of JavaScript's `Date.prototype.toISOString()`. Every value has the same width. So comparing two
+values as text gives the same order as comparing them as times. This matters for destinations that compare strings,
+such as a DynamoDB sort key or a CloudWatch Logs Insights filter.
+
+`emittedAt` strictly increases within one invocation: two records built in the same millisecond get values 1 ms
+apart, because the SQS FIFO deduplication id and the default DynamoDB sort key use `executionArn` and `emittedAt`, so
+equal values would drop or overwrite a record.
+
 ## Error names and causes
 
 Every error in a record (the execution `error`, each `operations[].error`, and each

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

@@ -9,12 +9,14 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongSupplier;
 import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
 import software.amazon.lambda.durable.plugin.InvocationEndInfo;
 import software.amazon.lambda.durable.plugin.InvocationInfo;
 import software.amazon.lambda.durable.plugin.OperationChangeInfo;
 import software.amazon.lambda.durable.plugin.OperationChangeItemInfo;
 import software.amazon.lambda.durable.telemetry.internal.FatalErrors;
+import software.amazon.lambda.durable.telemetry.internal.Timestamps;
 
 /**
  * The Durable Execution Telemetry plugin instance for one Lambda invocation: both the state that invocation's records
@@ -42,6 +44,8 @@
  *   <li><em>The build revision</em> — {@link #buildRevision} — counts the record builds this invocation has started, so
  *       that a build which was overtaken can be recognized at hand-off time and its record dropped. Atomic rather than
  *       {@code volatile}, because the case it exists for is two builds running at once. See the field.
+ *   <li><em>The last emittedAt</em> — {@link #lastEmittedMs} — is atomic for the same reason: two builds can run at
+ *       once. See {@link #nextEmittedAt()}.
  *   <li><em>Scheduling state</em> — {@link #record}, {@link #settled}, {@link #exporting}, {@link #drainWaiters} and
  *       {@link #closed} — is shared with the export pump and guarded by the monitor of {@link #scheduler}. One monitor
  *       for the whole environment, not one per invocation, so the {@code closed} check and the hand-off of a record are
@@ -108,6 +112,17 @@ final class TelemetryPlugin implements DurableExecutionPlugin {
      */
     private final AtomicLong buildRevision = new AtomicLong();
 
+    // --- emittedAt ordering. ---
+
+    /** Current time in epoch milliseconds. {@code System::currentTimeMillis} outside tests. */
+    private final LongSupplier clockMillis;
+
+    /**
+     * The last {@code emittedAt} this instance assigned, in epoch milliseconds. {@link Long#MIN_VALUE} before the first
+     * record. See {@link #nextEmittedAt()}.
+     */
+    private final AtomicLong lastEmittedMs = new AtomicLong(Long.MIN_VALUE);
+
     // --- Scheduling state: guarded by the scheduler's monitor. ---
 
     /**
@@ -167,7 +182,14 @@ final class TelemetryPlugin implements DurableExecutionPlugin {
      *     contains a hook failure, by skipping this plugin for the invocation
      */
     TelemetryPlugin(TelemetrySettings settings, ExportScheduler scheduler, InvocationInfo info) {
+        this(settings, scheduler, info, System::currentTimeMillis);
+    }
+
+    /** Test seam: the same instance with a supplied clock for {@code emittedAt}. */
+    TelemetryPlugin(
+            TelemetrySettings settings, ExportScheduler scheduler, InvocationInfo info, LongSupplier clockMillis) {
         this.settings = settings;
+        this.clockMillis = clockMillis;
         this.scheduler = scheduler;
         this.executionArn = info.durableExecutionArn();
         this.arn = ArnParser.parse(executionArn);
@@ -347,6 +369,30 @@ boolean isNewestBuild(long revision) {
         return buildRevision.get() == revision;
     }
 
+    /**
+     * Returns the {@code emittedAt} value for the next record this instance builds.
+     *
+     * <ol>
+     *   <li>{@code emittedAt} has millisecond precision.
+     *   <li>With {@code ON_CHANGE}, this instance can build two records in the same millisecond. One example is a
+     *       RUNNING record and the terminal record.
+     *   <li>The SQS FIFO deduplication id and the default DynamoDB sort key use {@code executionArn} and
+     *       {@code emittedAt}.
+     *   <li>So two records with equal values collide. SQS drops the second message. DynamoDB overwrites the first item.
+     *   <li>So {@code emittedAt} strictly increases within this instance. This instance serves one invocation.
+     * </ol>
+     *
+     * <p>The rule: if the clock is past the last value, the clock is used. Otherwise the last value plus 1 ms is used.
+     *
+     * <p>Hook threads can build records at the same time, and no lock serializes the builds. So the read and the update
+     * of the last value are one atomic {@code updateAndGet}. {@code Math.max(prev + 1, now)} is the rule above.
+     */
+    String nextEmittedAt() {
+        long nowMs = clockMillis.getAsLong();
+        long ms = lastEmittedMs.updateAndGet(prev -> Math.max(prev + 1, nowMs));
+        return Timestamps.format(Instant.ofEpochMilli(ms));
+    }
+
     private DurableExecutionTelemetryRecord buildRecord(
             String status,
             Map<String, OperationChangeItemInfo> operations,
@@ -356,17 +402,17 @@ private DurableExecutionTelemetryRecord buildRecord(
             Throwable error) {
         ContentConfig content = settings.content;
         DurableExecutionTelemetryRecord record = new DurableExecutionTelemetryRecord();
-        record.emittedAt = Instant.now().toString();
+        record.emittedAt = nextEmittedAt();
         record.executionArn = executionArn;
         record.executionName = DurableExecutionTelemetry.emptyToNull(arn.executionName());
         record.functionName = arn.functionName();
         record.functionQualifier = arn.qualifier();
         record.region = arn.region();
         record.accountId = arn.accountId();
         record.status = status;
-        record.startTime = startTime != null ? startTime.toString() : null;
+        record.startTime = startTime != null ? Timestamps.format(startTime) : null;
         if (endTime != null) {
-            record.endTime = endTime.toString();
+            record.endTime = Timestamps.format(endTime);
             if (startTime != null) {
                 record.durationMs = endTime.toEpochMilli() - startTime.toEpochMilli();
             }
@@ -429,11 +475,8 @@ private List<OperationRecord> buildOperationRecords(Map<String, OperationChangeI
                     .subType(item.subType())
                     .parentId(item.parentId())
                     .status(item.status() != null ? item.status().toString() : "UNKNOWN")
-                    .startTime(
-                            item.startTimestamp() != null
-                                    ? item.startTimestamp().toString()
-                                    : null)
-                    .endTime(item.endTimestamp() != null ? item.endTimestamp().toString() : null)
+                    .startTime(item.startTimestamp() != null ? Timestamps.format(item.startTimestamp()) : null)
+                    .endTime(item.endTimestamp() != null ? Timestamps.format(item.endTimestamp()) : null)
                     .attempt(item.attempt());
             if (item.startTimestamp() != null && item.endTimestamp() != null) {
                 rec.durationMs(item.endTimestamp().toEpochMilli()

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

@@ -3,13 +3,12 @@
 package software.amazon.lambda.durable.telemetry.exporters;
 
 import java.time.Instant;
-import java.time.ZoneOffset;
-import java.time.format.DateTimeFormatter;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetryRecord;
 import software.amazon.lambda.durable.telemetry.Json;
 import software.amazon.lambda.durable.telemetry.TelemetryExporter;
+import software.amazon.lambda.durable.telemetry.internal.Timestamps;
 
 /**
  * Exports workflow telemetry records to CloudWatch Logs via {@code System.out.println} (Lambda captures stdout to the
@@ -35,10 +34,6 @@
  * or above, these lines are dropped, as raw record lines were before.
  */
 public final class LambdaLogExporter implements TelemetryExporter {
-    /** UTC ISO-8601 with exactly three fraction digits. */
-    private static final DateTimeFormatter TIMESTAMP =
-            DateTimeFormatter.ofPattern("uuuu-MM-dd'T'HH:mm:ss.SSS'Z'").withZone(ZoneOffset.UTC);
-
     /** UTF-8 size of the envelope around an empty {@code message}; the timestamp always has the same width. */
     private static final int ENVELOPE_BYTES =
             logLine("", Instant.EPOCH).getBytes(java.nio.charset.StandardCharsets.UTF_8).length;
@@ -101,7 +96,7 @@ static String logLine(String recordJson, Instant now) {
 
     private static Map<String, Object> envelope(String recordJson, Instant now) {
         Map<String, Object> line = new LinkedHashMap<>();
-        line.put("timestamp", TIMESTAMP.format(now));
+        line.put("timestamp", Timestamps.format(now));
         line.put("level", "INFO");
         line.put("message", recordJson);
         return line;

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

@@ -0,0 +1,31 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.telemetry.internal;
+
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+
+/** Internal timestamp format for Telemetry records and the Lambda log envelope. */
+public final class Timestamps {
+    /** UTC ISO-8601 with exactly three fraction digits and a literal {@code Z}. */
+    private static final DateTimeFormatter FORMAT =
+            DateTimeFormatter.ofPattern("uuuu-MM-dd'T'HH:mm:ss.SSS'Z'").withZone(ZoneOffset.UTC);
+
+    private Timestamps() {}
+
+    /**
+     * Formats {@code instant} as {@code YYYY-MM-DDTHH:MM:SS.sssZ}, the format of JavaScript's
+     * {@code Date.prototype.toISOString()}.
+     *
+     * <p>The fraction always has three digits. A zero fraction is written as {@code .000}. Digits below the millisecond
+     * are dropped, not rounded, so {@code .078757096} becomes {@code .078} and {@code .999999999} becomes {@code .999}.
+     *
+     * <p>Destinations such as DynamoDB and CloudWatch Logs Insights compare these values as text. Text order equals
+     * time order only when every value has the same width. {@link Instant#toString()} writes 0, 3, 6 or 9 fraction
+     * digits. So every record timestamp goes through this method instead.
+     */
+    public static String format(Instant instant) {
+        return FORMAT.format(instant);
+    }
+}

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

@@ -39,7 +39,7 @@ private DurableExecutionTelemetryRecord sampleRecord() {
         DurableExecutionTelemetryRecord r = new DurableExecutionTelemetryRecord();
         r.executionArn = "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1";
         r.status = "SUCCEEDED";
-        r.startTime = "2026-08-05T00:00:00Z";
+        r.startTime = "2026-08-05T00:00:00.000Z";
         r.addOperation(new OperationRecord()
                 .id("op-1")
                 .name("greet")

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

@@ -265,7 +265,7 @@ void suspendResumeKeepsStableStartTimeAndLeavesNoRetainedState() {
         assertEquals(4, exporter.records.size());
         String startTime = exporter.records.get(0).startTime();
         assertTrue(exporter.records.stream().allMatch(r -> startTime.equals(r.startTime())));
-        assertEquals(START.toString(), startTime);
+        assertEquals("2026-08-05T00:00:00.000Z", startTime);
         assertFalse(Executions.outstanding(first), "the suspended invocation left the scheduler owing nothing");
         assertFalse(Executions.outstanding(resumed), "nor did the resumed one");
     }

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

@@ -0,0 +1,152 @@
+// 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 java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.jupiter.api.Test;
+import software.amazon.lambda.durable.plugin.InvocationEndInfo;
+import software.amazon.lambda.durable.plugin.InvocationInfo;
+import software.amazon.lambda.durable.plugin.InvocationStatus;
+import software.amazon.lambda.durable.plugin.OperationChangeInfo;
+
+/**
+ * {@code emittedAt} strictly increases within one plugin instance.
+ *
+ * <p>{@code emittedAt} has millisecond precision. With {@code ON_CHANGE}, two records can be built in the same
+ * millisecond. The SQS FIFO deduplication id and the default DynamoDB sort key use {@code executionArn} and
+ * {@code emittedAt}. So two equal values drop or overwrite a record.
+ */
+class EmittedAtOrderingTest {
+
+    private static final String ARN =
+            "arn:aws:lambda:us-west-2:111122223333:function:f:$LATEST/durable-execution/exec-1/invocation-1";
+    private static final Instant START = Instant.parse("2026-10-23T18:42:40Z");
+    private static final long T123 = Instant.parse("2026-10-23T18:42:44.123Z").toEpochMilli();
+
+    private static final class Fixture {
+        final List<DurableExecutionTelemetryRecord> exported = Collections.synchronizedList(new ArrayList<>());
+        final AtomicLong clock = new AtomicLong(T123);
+        final TelemetryPlugin plugin;
+
+        Fixture() {
+            var config = DurableExecutionTelemetryConfig.builder()
+                    .emitMode(DurableExecutionTelemetryConfig.EmitMode.ON_CHANGE)
+                    .addExporter(exported::add)
+                    .build();
+            // Runnable::run runs the pump inside each hand-off. So every record reaches the exporter before the next
+            // record is built, and the scheduler coalesces none of them.
+            var scheduler = new ExportScheduler(
+                    config.exporters(), (record, target) -> target.export(record), t -> {}, Runnable::run);
+            plugin = new TelemetryPlugin(
+                    new TelemetrySettings(config),
+                    scheduler,
+                    new InvocationInfo("req", ARN, true, START, null, Map.of(), Map.of()),
+                    clock::get);
+        }
+
+        void start() {
+            plugin.onInvocationStart(new InvocationInfo("req", ARN, true, START, "in", Map.of(), Map.of()));
+        }
+
+        void change() {
+            plugin.onOperationChange(new OperationChangeInfo("req", ARN, Map.of(), Map.of()));
+        }
+
+        void end() {
+            plugin.onInvocationEnd(new InvocationEndInfo(
+                    "req", ARN, true, START, Map.of(), InvocationStatus.SUCCEEDED, null, "in", "out"));
+        }
+
+        List<String> statusAndEmittedAt() {
+            List<String> out = new ArrayList<>();
+            synchronized (exported) {
+                for (var record : exported) {
+                    out.add(record.status() + " " + record.emittedAt);
+                }
+            }
+            return out;
+        }
+    }
+
+    @Test
+    void recordsBuiltInOneMillisecondGetIncreasingValues() {
+        var fixture = new Fixture();
+
+        fixture.start();
+        fixture.change();
+        fixture.end();
+
+        assertEquals(
+                List.of(
+                        "RUNNING 2026-10-23T18:42:44.123Z",
+                        "RUNNING 2026-10-23T18:42:44.124Z",
+                        "SUCCEEDED 2026-10-23T18:42:44.125Z"),
+                fixture.statusAndEmittedAt());
+    }
+
+    @Test
+    void theClockIsUsedOnceItPassesTheLastValue() {
+        var fixture = new Fixture();
+
+        fixture.start();
+        fixture.clock.set(T123 + 377);
+        fixture.change();
+        // The clock moves backwards. The last value plus 1 ms is used.
+        fixture.clock.set(T123 + 100);
+        fixture.end();
+
+        assertEquals(
+                List.of(
+                        "RUNNING 2026-10-23T18:42:44.123Z",
+                        "RUNNING 2026-10-23T18:42:44.500Z",
+                        "SUCCEEDED 2026-10-23T18:42:44.501Z"),
+                fixture.statusAndEmittedAt());
+    }
+
+    @Test
+    void concurrentCallsGetDistinctValues() throws Exception {
+        var fixture = new Fixture();
+        int threads = 8;
+        int callsPerThread = 2_000;
+        var ready = new CountDownLatch(threads);
+        var go = new CountDownLatch(1);
+        ExecutorService pool = Executors.newFixedThreadPool(threads);
+        try {
+            List<Future<List<String>>> futures = new ArrayList<>();
+            for (int t = 0; t < threads; t++) {
+                futures.add(pool.submit(() -> {
+                    List<String> values = new ArrayList<>(callsPerThread);
+                    ready.countDown();
+                    go.await();
+                    for (int i = 0; i < callsPerThread; i++) {
+                        values.add(fixture.plugin.nextEmittedAt());
+                    }
+                    return values;
+                }));
+            }
+            ready.await();
+            go.countDown();
+            Set<String> all = new HashSet<>();
+            for (var future : futures) {
+                all.addAll(future.get(30, TimeUnit.SECONDS));
+            }
+            assertEquals(threads * callsPerThread, all.size(), "every emittedAt must be distinct");
+        } finally {
+            pool.shutdownNow();
+        }
+    }
+}