Fixed test server previous started eventId (#136)

Author: mfateev

src/main/java/com/uber/cadence/internal/replay/HistoryHelper.java

@@ -112,7 +112,7 @@ public PollForDecisionTaskResponse getDecisionTask() {
     return events.getDecisionTask();
   }
 
-  public long getLastNonReplayEventId() {
+  public long getPreviousStartedEventId() {
     return getDecisionTask().getPreviousStartedEventId();
   }
 }

src/main/java/com/uber/cadence/internal/replay/ReplayDecider.java

@@ -343,7 +343,8 @@ public void decide() throws Throwable {
 
   private void decideImpl(Functions.Proc query) throws Throwable {
     try {
-      long lastNonReplayedEventId = historyHelper.getLastNonReplayEventId();
+      long previousDecisionStartedEventId = historyHelper.getPreviousStartedEventId();
+      long lastNonFailedStartEventId = -1;
       // Buffer events until the next DecisionTaskStarted and then process them
       // setting current time to the time of DecisionTaskStarted event
       HistoryHelper.EventsIterator eventsIterator = historyHelper.getEvents();
@@ -367,6 +368,12 @@ private void decideImpl(Functions.Proc query) throws Throwable {
 
             decisionsHelper.handleDecisionTaskStartedEvent();
             if (!eventsIterator.isNextDecisionFailed()) {
+              if (query == null
+                  && lastNonFailedStartEventId > 0
+                  && previousDecisionStartedEventId == 0) {
+                throw new Error("Unexpected 0 previousDecisionStartedEventId: ");
+              }
+              lastNonFailedStartEventId = event.getEventId();
               // Cadence timestamp is in nanoseconds
               long replayCurrentTimeMilliseconds = event.getTimestamp() / MILLION;
               context.setReplayCurrentTimeMilliseconds(replayCurrentTimeMilliseconds);
@@ -379,7 +386,7 @@ private void decideImpl(Functions.Proc query) throws Throwable {
           }
         }
         for (HistoryEvent event : decisionCompletionToStartEvents) {
-          if (event.getEventId() >= lastNonReplayedEventId) {
+          if (event.getEventId() >= previousDecisionStartedEventId) {
             context.setReplaying(false);
           }
           EventType eventType = event.getEventType();

src/main/java/com/uber/cadence/internal/testservice/StateMachines.java

@@ -143,17 +143,20 @@ static final class WorkflowData {}
 
   static final class DecisionTaskData {
 
+    final long previousStartedEventId;
+
     final TestWorkflowStore store;
 
-    long previousStartedEventId = -1;
+    long startedEventId = -1;
 
     PollForDecisionTaskResponse decisionTask;
 
     long scheduledEventId = -1;
 
     int attempt;
 
-    DecisionTaskData(TestWorkflowStore store) {
+    DecisionTaskData(long previousStartedEventId, TestWorkflowStore store) {
+      this.previousStartedEventId = previousStartedEventId;
       this.store = store;
     }
   }
@@ -212,8 +215,9 @@ static StateMachine<WorkflowData> newWorkflowStateMachine() {
         .add(CANCELLATION_REQUESTED, TIME_OUT, TIMED_OUT, StateMachines::timeoutWorkflow);
   }
 
-  static StateMachine<DecisionTaskData> newDecisionStateMachine(TestWorkflowStore store) {
-    return new StateMachine<>(new DecisionTaskData(store))
+  static StateMachine<DecisionTaskData> newDecisionStateMachine(
+      long previousStartedEventId, TestWorkflowStore store) {
+    return new StateMachine<>(new DecisionTaskData(previousStartedEventId, store))
         .add(NONE, INITIATE, INITIATED, StateMachines::scheduleDecisionTask)
         .add(INITIATED, START, STARTED, StateMachines::startDecisionTask)
         .add(STARTED, COMPLETE, COMPLETED, StateMachines::completeDecisionTask)
@@ -694,7 +698,7 @@ private static void startDecisionTask(
             throw new InternalServiceError(entityNotExistsError.toString());
           }
           data.decisionTask.setHistory(new History().setEvents(events));
-          data.previousStartedEventId = startedEventId;
+          data.startedEventId = startedEventId;
           data.attempt++;
         });
   }
@@ -746,7 +750,7 @@ private static void failDecisionTask(
             .setIdentity(request.getIdentity())
             .setCause(request.getCause())
             .setDetails(request.getDetails())
-            .setStartedEventId(data.previousStartedEventId)
+            .setStartedEventId(data.startedEventId)
             .setScheduledEventId(data.scheduledEventId);
     HistoryEvent event =
         new HistoryEvent()
@@ -759,7 +763,7 @@ private static void timeoutDecisionTask(
       RequestContext ctx, DecisionTaskData data, Object ignored, long notUsed) {
     DecisionTaskTimedOutEventAttributes a =
         new DecisionTaskTimedOutEventAttributes()
-            .setStartedEventId(data.previousStartedEventId)
+            .setStartedEventId(data.startedEventId)
             .setTimeoutType(TimeoutType.START_TO_CLOSE)
             .setScheduledEventId(data.scheduledEventId);
     HistoryEvent event =

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowMutableStateImpl.java

@@ -129,6 +129,7 @@ void apply(RequestContext ctx)
   private final Map<String, StateMachine<SignalExternalData>> externalSignals = new HashMap<>();
   private StateMachine<WorkflowData> workflow;
   private StateMachine<DecisionTaskData> decision;
+  private long lastNonFailedDecisionStartEventId;
   private final Map<String, CompletableFuture<QueryWorkflowResponse>> queries =
       new ConcurrentHashMap<>();
 
@@ -252,6 +253,7 @@ public void completeDecisionTask(int historySize, RespondDecisionTaskCompletedRe
           for (RequestContext deferredCtx : this.concurrentToDecision) {
             ctx.add(deferredCtx);
           }
+          lastNonFailedDecisionStartEventId = this.decision.getData().startedEventId;
           this.decision = null;
           boolean completed =
               workflow.getState() == StateMachines.State.COMPLETED
@@ -823,7 +825,7 @@ private void scheduleDecision(RequestContext ctx) throws InternalServiceError {
       }
       throw new InternalServiceError("unexpected decision state: " + decision.getState());
     }
-    this.decision = StateMachines.newDecisionStateMachine(store);
+    this.decision = StateMachines.newDecisionStateMachine(lastNonFailedDecisionStartEventId, store);
     decision.action(StateMachines.Action.INITIATE, ctx, startRequest, 0);
     ctx.lockTimer();
   }