fix infinite loop when use continueasnew after wait external event (#183)

Author: kaibocai

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

@@ -787,12 +787,17 @@ private void addCarryoverEvents(CompleteOrchestrationAction.Builder builder) {
             // We don't check the event in the pass event list to avoid duplicated events.
             Set<HistoryEvent> externalEvents = new HashSet<>(this.unprocessedEvents);
             List<HistoryEvent> newEvents = this.historyEventPlayer.getNewEvents();
+            int currentHistoryIndex = this.historyEventPlayer.getCurrentHistoryIndex();
+
+            // Only add events that haven't been processed to the carryOverEvents
+            // currentHistoryIndex will point to the first unprocessed event
+            for (int i = currentHistoryIndex; i < newEvents.size(); i++) {
+                HistoryEvent historyEvent = newEvents.get(i);
+                if (historyEvent.getEventTypeCase() == HistoryEvent.EventTypeCase.EVENTRAISED) {
+                    externalEvents.add(historyEvent);
+                }
+            }
 
-            Set<HistoryEvent> filteredEvents = newEvents.stream()
-                    .filter(e -> e.getEventTypeCase() == HistoryEvent.EventTypeCase.EVENTRAISED)
-                    .collect(Collectors.toSet());
-
-            externalEvents.addAll(filteredEvents);
             externalEvents.forEach(builder::addCarryoverEvents);
         }
 
@@ -946,7 +951,11 @@ public boolean moveNext() {
             }
 
             List<HistoryEvent> getNewEvents() {
-                return newEvents;
+                return this.newEvents;
+            }
+
+            int getCurrentHistoryIndex() {
+                return this.currentHistoryIndex;
             }
         }
 

samples-azure-functions/src/main/java/com/functions/ContinueAsNew.java

@@ -8,6 +8,7 @@
 import com.microsoft.azure.functions.annotation.FunctionName;
 import com.microsoft.azure.functions.annotation.HttpTrigger;
 import com.microsoft.durabletask.DurableTaskClient;
+import com.microsoft.durabletask.Task;
 import com.microsoft.durabletask.TaskOrchestrationContext;
 import com.microsoft.durabletask.azurefunctions.DurableClientContext;
 import com.microsoft.durabletask.azurefunctions.DurableClientInput;
@@ -37,4 +38,29 @@ public void eternalOrchestrator(@DurableOrchestrationTrigger(name = "runtimeStat
         ctx.createTimer(Duration.ofSeconds(2)).await();
         ctx.continueAsNew(null);
     }
+
+    @FunctionName("ContinueAsNewExternalEvent")
+    public HttpResponseMessage continueAsNewExternalEvent(
+            @HttpTrigger(name = "req", methods = {HttpMethod.GET, HttpMethod.POST}, authLevel = AuthorizationLevel.ANONYMOUS) HttpRequestMessage<Optional<String>> request,
+            @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
+            final ExecutionContext context) {
+        context.getLogger().info("Java HTTP trigger processed a request.");
+
+        DurableTaskClient client = durableContext.getClient();
+        String instanceId = client.scheduleNewOrchestrationInstance("EternalEvent");
+        context.getLogger().info("Created new Java orchestration with instance ID = " + instanceId);
+        return durableContext.createCheckStatusResponse(request, instanceId);
+    }
+
+    @FunctionName("EternalEvent")
+    public void eternalEvent(@DurableOrchestrationTrigger(name = "runtimeState") TaskOrchestrationContext ctx)
+    {
+        System.out.println("Waiting external event...");
+        Task<Void> event = ctx.waitForExternalEvent("event");
+        Task<Void> timer = ctx.createTimer(Duration.ofSeconds(10));
+        Task<?> result = ctx.anyOf(event, timer).await();
+        if (result == event) {
+            ctx.continueAsNew(null);
+        }
+    }
 }

samples-azure-functions/src/test/java/com/functions/EndToEndTests.java

@@ -14,6 +14,7 @@
 import java.time.Instant;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import static io.restassured.RestAssured.get;
 import static io.restassured.RestAssured.post;
@@ -99,6 +100,37 @@ public void continueAsNew() throws InterruptedException {
         statusResponse.jsonPath().get("runtimeStatus");
     }
 
+    @Test
+    public void continueAsNewExternalEvent() throws InterruptedException {
+        String startOrchestrationPath = "api/ContinueAsNewExternalEvent";
+        Response response = post(startOrchestrationPath);
+        JsonPath jsonPath = response.jsonPath();
+
+        // send external event, it will cause the continue-as-new.
+        String sendEventPostUri = jsonPath.get("sendEventPostUri");
+        sendEventPostUri = sendEventPostUri.replace("{eventName}", "event");
+
+        // empty request body
+        RestAssured
+                .given()
+                .contentType(ContentType.JSON) // Set the request content type
+                .body("{}")
+                .post(sendEventPostUri)
+                .then()
+                .statusCode(202);
+
+        //wait 5 seconds for the continue-as-new to start new orchestration
+        TimeUnit.SECONDS.sleep(5);
+
+        response = post(startOrchestrationPath);
+        jsonPath = response.jsonPath();
+        String statusQueryGetUri = jsonPath.get("statusQueryGetUri");
+        // polling 20 seconds
+        // assert that the orchestration completed as expected, not enter an infinite loop
+        boolean completed = pollingCheck(statusQueryGetUri, "Completed", null, Duration.ofSeconds(20));
+        assertTrue(completed);
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void restart(boolean restartWithNewInstanceId) throws InterruptedException {