addressed PR feedback

Author: bachuv

client/src/main/java/com/microsoft/durabletask/OrchestrationHistoryEventMapper.java

@@ -22,6 +22,8 @@
  */
 final class OrchestrationHistoryEventMapper {
 
+    private static final int MAX_PROTO_NESTING_DEPTH = 10;
+
     private static final Map<HistoryEvent.EventTypeCase, String> EVENT_TYPE_NAMES = new HashMap<>();
 
     static {
@@ -104,9 +106,9 @@ private static Map<String, Object> extractEventData(
             // Proto map fields appear as repeated MapEntry messages in reflection;
             // convert them to plain Java Maps using the field descriptor.
             if (field.isMapField()) {
-                data.put(field.getJsonName(), convertProtoMapField(field, value));
+                data.put(field.getJsonName(), convertProtoMapField(field, value, 0));
             } else {
-                data.put(field.getJsonName(), convertProtoValue(value));
+                data.put(field.getJsonName(), convertProtoValue(value, 0));
             }
         }
 
@@ -148,7 +150,7 @@ private static Message getEventMessage(HistoryEvent proto, HistoryEvent.EventTyp
     }
 
     @SuppressWarnings("unchecked")
-    private static Map<String, Object> convertProtoMapField(Descriptors.FieldDescriptor field, Object value) {
+    private static Map<String, Object> convertProtoMapField(Descriptors.FieldDescriptor field, Object value, int depth) {
         // Proto map fields are represented as List<MapEntry> in reflection.
         // Each MapEntry is a Message with "key" and "value" fields.
         List<Message> entries = (List<Message>) value;
@@ -157,13 +159,16 @@ private static Map<String, Object> convertProtoMapField(Descriptors.FieldDescrip
         Descriptors.FieldDescriptor valueField = field.getMessageType().findFieldByName("value");
         for (Message entry : entries) {
             String key = String.valueOf(entry.getField(keyField));
-            Object val = convertProtoValue(entry.getField(valueField));
+            Object val = convertProtoValue(entry.getField(valueField), depth + 1);
             result.put(key, val);
         }
         return result;
     }
 
-    private static Object convertProtoValue(Object value) {
+    private static Object convertProtoValue(Object value, int depth) {
+        if (depth > MAX_PROTO_NESTING_DEPTH) {
+            return "[nested object truncated at max depth " + MAX_PROTO_NESTING_DEPTH + "]";
+        }
         if (value instanceof Timestamp) {
             Timestamp ts = (Timestamp) value;
             return toInstant(ts).toString();
@@ -182,7 +187,7 @@ private static Object convertProtoValue(Object value) {
             List<?> protoList = (List<?>) value;
             List<Object> converted = new ArrayList<>(protoList.size());
             for (Object item : protoList) {
-                converted.add(convertProtoValue(item));
+                converted.add(convertProtoValue(item, depth + 1));
             }
             return converted;
         }
@@ -191,7 +196,7 @@ private static Object convertProtoValue(Object value) {
             Map<String, Object> nested = new LinkedHashMap<>();
             for (Map.Entry<Descriptors.FieldDescriptor, Object> entry :
                     ((Message) value).getAllFields().entrySet()) {
-                nested.put(entry.getKey().getJsonName(), convertProtoValue(entry.getValue()));
+                nested.put(entry.getKey().getJsonName(), convertProtoValue(entry.getValue(), depth + 1));
             }
             return nested;
         }

exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/client/DefaultExportHistoryJobClient.java

@@ -169,20 +169,34 @@ private void terminateAndPurgeOrchestration(String orchestrationInstanceId) {
     /**
      * Detects "instance not found" responses from the sidecar without taking a hard dependency on
      * gRPC types. The Durable Task gRPC client surfaces "not found" as a {@code StatusRuntimeException}
-     * with code {@code NOT_FOUND}; we match against the exception's {@code toString()} which embeds
-     * both. This mirrors .NET's {@code catch (RpcException ex) when (ex.StatusCode == StatusCode.NotFound)}.
+     * with code {@code NOT_FOUND}; we check the status code via reflection for exactness. Mirrors
+     * .NET's {@code catch (RpcException ex) when (ex.StatusCode == StatusCode.NotFound)}.
      */
     private static boolean isNotFound(Throwable t) {
         while (t != null) {
             String name = t.getClass().getName();
             if ("io.grpc.StatusRuntimeException".equals(name) || "io.grpc.StatusException".equals(name)) {
-                String msg = String.valueOf(t.getMessage());
-                if (msg.contains("NOT_FOUND")) {
+                if (hasGrpcStatusCode(t, "NOT_FOUND")) {
                     return true;
                 }
             }
             t = t.getCause();
         }
         return false;
     }
+
+    /**
+     * Extracts the gRPC status code name via reflection and compares it to {@code expected}.
+     * Falls back to a {@code "CODE: "} message prefix check if reflection is unavailable.
+     */
+    private static boolean hasGrpcStatusCode(Throwable t, String expected) {
+        try {
+            Object status = t.getClass().getMethod("getStatus").invoke(t);
+            Object code = status.getClass().getMethod("getCode").invoke(status);
+            return expected.equals(code.toString());
+        } catch (ReflectiveOperationException ignored) {
+            // Fallback: gRPC formats messages as "CODE: description"
+            return String.valueOf(t.getMessage()).startsWith(expected + ":");
+        }
+    }
 }

exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/models/ExportJobCreationOptions.java

@@ -55,7 +55,7 @@ public ExportJobCreationOptions() {
      * @param jobId               the job ID, or {@code null} to auto-generate
      * @param mode                the export mode (BATCH or CONTINUOUS)
      * @param completedTimeFrom   inclusive start of the completed time window
-     * @param completedTimeTo     exclusive end of the completed time window — must be strictly after
+     * @param completedTimeTo     inclusive end of the completed time window — must be strictly after
      *                            {@code completedTimeFrom} (required for BATCH, null for CONTINUOUS)
      * @param destination         blob storage destination, or {@code null} to use defaults
      * @param format              export format, or {@code null} for default (JSONL+gzip)

exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/orchestrations/ExportJobOrchestrator.java

@@ -3,6 +3,7 @@
 package com.microsoft.durabletask.exporthistory.orchestrations;
 
 import com.microsoft.durabletask.Task;
+import com.microsoft.durabletask.TaskFailedException;
 import com.microsoft.durabletask.TaskOptions;
 import com.microsoft.durabletask.TaskOrchestration;
 import com.microsoft.durabletask.TaskOrchestrationContext;
@@ -129,14 +130,19 @@ public void run(TaskOrchestrationContext ctx) {
             // SDK control-flow signals — must propagate so the runtime can suspend/continue.
             throw controlFlow;
         } catch (Exception ex) {
-            // Mark as failed
+            // Mark as failed (best-effort). The inner try MUST re-throw SDK control-flow signals;
+            // otherwise the entity call's await() block on its first turn will be swallowed and the
+            // orchestrator will fall through to `throw ex` in the same turn, leaving the ExportJob
+            // entity stuck in ACTIVE and breaking replay determinism.
             try {
                 ctx.callEntity(input.getJobEntityId(),
                         ExportJobOperationNames.MARK_AS_FAILED,
                         ex.getMessage(),
                         Void.class).await();
+            } catch (OrchestratorBlockedException | ContinueAsNewInterruption controlFlow) {
+                throw controlFlow;
             } catch (Exception ignored) {
-                // Best-effort failure marking
+                // Best-effort failure marking — swallow only genuine application errors
             }
             throw ex instanceof RuntimeException ? (RuntimeException) ex : new RuntimeException(ex);
         }
@@ -213,11 +219,16 @@ List<ExportResult> exportBatch(
                         ExportResult.class));
             }
 
-            // Wait for all exports in this chunk before scheduling the next
+            // Wait for all exports in this chunk before scheduling the next.
+            // IMPORTANT: only catch TaskFailedException — never a broad Exception. The SDK's
+            // OrchestratorBlockedException / ContinueAsNewInterruption are control-flow signals
+            // that MUST propagate so the runtime can suspend/continue the orchestration. Swallowing
+            // them produces non-deterministic replay (e.g. fabricated failures + spurious timers in
+            // history that don't match the replayed action sequence).
             for (int i = 0; i < exportTasks.size(); i++) {
                 try {
                     results.add(exportTasks.get(i).await());
-                } catch (Exception ex) {
+                } catch (TaskFailedException ex) {
                     // Activity failure after all retries — preserve the instance ID for diagnostics
                     results.add(new ExportResult(chunk.get(i), false, ex.getMessage()));
                 }

exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/orchestrations/ExportJobOrchestratorChunkingTest.java

@@ -3,8 +3,11 @@
 package com.microsoft.durabletask.exporthistory.orchestrations;
 
 import com.microsoft.durabletask.Task;
+import com.microsoft.durabletask.TaskFailedException;
 import com.microsoft.durabletask.TaskOptions;
 import com.microsoft.durabletask.TaskOrchestrationContext;
+import com.microsoft.durabletask.interruption.ContinueAsNewInterruption;
+import com.microsoft.durabletask.interruption.OrchestratorBlockedException;
 import com.microsoft.durabletask.exporthistory.models.ExportDestination;
 import com.microsoft.durabletask.exporthistory.models.ExportFilter;
 import com.microsoft.durabletask.exporthistory.models.ExportFormat;
@@ -156,7 +159,9 @@ void exportBatch_activityFailure_capturesFailureWithInstanceId() {
                     @SuppressWarnings("unchecked")
                     Task<ExportResult> task = mock(Task.class);
                     if ("fail-me".equals(req.getInstanceId())) {
-                        when(task.await()).thenThrow(new RuntimeException("activity exhausted retries"));
+                        TaskFailedException tfe = mock(TaskFailedException.class);
+                        when(tfe.getMessage()).thenReturn("activity exhausted retries");
+                        when(task.await()).thenThrow(tfe);
                     } else {
                         when(task.await()).thenReturn(new ExportResult(req.getInstanceId(), true, null));
                     }
@@ -185,4 +190,71 @@ void exportBatch_activityFailure_capturesFailureWithInstanceId() {
         assertTrue(results.get(1).getError().contains("activity exhausted retries"));
         assertTrue(results.get(2).isSuccess());
     }
+
+    /**
+     * A broad {@code catch (Exception)} around {@code Task.await()} would swallow
+     * {@link OrchestratorBlockedException}, fabricating a "failed" {@link ExportResult} on the
+     * first orchestrator turn and producing non-deterministic replay. The catch must be narrowed
+     * to {@link TaskFailedException} so SDK control-flow signals propagate.
+     */
+    @Test
+    void exportBatch_orchestratorBlockedException_propagates() {
+        TaskOrchestrationContext ctx = mock(TaskOrchestrationContext.class);
+        when(ctx.callActivity(eq("ExportInstanceHistoryActivity"), any(ExportRequest.class),
+                any(TaskOptions.class), eq(ExportResult.class)))
+                .thenAnswer((InvocationOnMock inv) -> {
+                    @SuppressWarnings("unchecked")
+                    Task<ExportResult> task = mock(Task.class);
+                    when(task.await()).thenThrow(
+                            new OrchestratorBlockedException("awaiting incomplete task"));
+                    return task;
+                });
+
+        List<String> instanceIds = new ArrayList<>();
+        instanceIds.add("i-1");
+        instanceIds.add("i-2");
+
+        ExportJobConfiguration config = new ExportJobConfiguration(
+                ExportMode.BATCH,
+                new ExportFilter(),
+                new ExportDestination("test-container", null),
+                ExportFormat.DEFAULT,
+                32,
+                100);
+
+        assertThrows(OrchestratorBlockedException.class,
+                () -> new ExportJobOrchestrator().exportBatch(ctx, instanceIds, config),
+                "OrchestratorBlockedException must propagate so the runtime can suspend");
+    }
+
+    /**
+     * Companion to the OrchestratorBlockedException test — {@link ContinueAsNewInterruption}
+     * is the other SDK control-flow signal that must never be swallowed by user code.
+     */
+    @Test
+    void exportBatch_continueAsNewInterruption_propagates() {
+        TaskOrchestrationContext ctx = mock(TaskOrchestrationContext.class);
+        when(ctx.callActivity(eq("ExportInstanceHistoryActivity"), any(ExportRequest.class),
+                any(TaskOptions.class), eq(ExportResult.class)))
+                .thenAnswer((InvocationOnMock inv) -> {
+                    @SuppressWarnings("unchecked")
+                    Task<ExportResult> task = mock(Task.class);
+                    when(task.await()).thenThrow(new ContinueAsNewInterruption("continue-as-new"));
+                    return task;
+                });
+
+        List<String> instanceIds = new ArrayList<>();
+        instanceIds.add("i-1");
+
+        ExportJobConfiguration config = new ExportJobConfiguration(
+                ExportMode.BATCH,
+                new ExportFilter(),
+                new ExportDestination("test-container", null),
+                ExportFormat.DEFAULT,
+                32,
+                100);
+
+        assertThrows(ContinueAsNewInterruption.class,
+                () -> new ExportJobOrchestrator().exportBatch(ctx, instanceIds, config));
+    }
 }