fix bugs that cause time skipping to not work in certain cases (#162)

* fix bugs that cause time skipping to not work in certain cases

* address review comments

* reduce timeout for WorkflowTest

Author: meiliang86

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

@@ -106,10 +106,15 @@ void unlockTimer() {
     timerLocks--;
   }
 
-  public int getTimerLocks() {
+  int getTimerLocks() {
     return timerLocks;
   }
 
+  void clearTimersAndLocks() {
+    timerLocks = 0;
+    timers.clear();
+  }
+
   long currentTimeInNanoseconds() {
     return clock.getAsLong() * NANOS_PER_MILLIS;
   }

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

@@ -133,7 +133,7 @@ void apply(RequestContext ctx)
   private final Map<String, StateMachine<TimerData>> timers = new HashMap<>();
   private final Map<String, StateMachine<SignalExternalData>> externalSignals = new HashMap<>();
   private StateMachine<WorkflowData> workflow;
-  private StateMachine<DecisionTaskData> decision;
+  private volatile StateMachine<DecisionTaskData> decision;
   private long lastNonFailedDecisionStartEventId;
   private final Map<String, CompletableFuture<QueryWorkflowResponse>> queries =
       new ConcurrentHashMap<>();
@@ -182,6 +182,7 @@ private void update(boolean completeDecisionUpdate, UpdateProcedure updater)
       if (concurrentDecision && workflow.getState() != State.TIMED_OUT) {
         concurrentToDecision.add(ctx);
         ctx.fireCallbacks(0);
+        store.applyTimersAndLocks(ctx);
       } else {
         nextEventId = ctx.commitChanges(store);
       }
@@ -606,6 +607,7 @@ public void childWorklfowFailed(String activityId, ChildWorkflowExecutionFailedE
           child.action(StateMachines.Action.FAIL, ctx, a, 0);
           childWorkflows.remove(a.getInitiatedEventId());
           scheduleDecision(ctx);
+          ctx.unlockTimer();
         });
   }
 
@@ -619,6 +621,7 @@ public void childWorklfowTimedOut(
           child.action(Action.TIME_OUT, ctx, a.getTimeoutType(), 0);
           childWorkflows.remove(a.getInitiatedEventId());
           scheduleDecision(ctx);
+          ctx.unlockTimer();
         });
   }
 
@@ -646,6 +649,7 @@ public void childWorkflowCompleted(
           child.action(StateMachines.Action.COMPLETE, ctx, a, 0);
           childWorkflows.remove(a.getInitiatedEventId());
           scheduleDecision(ctx);
+          ctx.unlockTimer();
         });
   }
 
@@ -659,6 +663,7 @@ public void childWorkflowCanceled(
           child.action(StateMachines.Action.CANCEL, ctx, a, 0);
           childWorkflows.remove(a.getInitiatedEventId());
           scheduleDecision(ctx);
+          ctx.unlockTimer();
         });
   }
 

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

@@ -137,6 +137,8 @@ public PollForActivityTaskResponse getTask() {
 
   long save(RequestContext requestContext) throws InternalServiceError, EntityNotExistsError;
 
+  void applyTimersAndLocks(RequestContext ctx);
+
   void registerDelayedCallback(Duration delay, Runnable r);
 
   PollForDecisionTaskResponse pollForDecisionTask(PollForDecisionTaskRequest pollRequest)

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

@@ -219,6 +219,25 @@ public long save(RequestContext ctx) throws InternalServiceError, EntityNotExist
     return result;
   }
 
+  @Override
+  public void applyTimersAndLocks(RequestContext ctx) {
+    lock.lock();
+    try {
+      timerService.updateLocks(ctx.getTimerLocks());
+    } finally {
+      lock.unlock();
+    }
+
+    List<Timer> timers = ctx.getTimers();
+    if (timers != null) {
+      for (Timer t : timers) {
+        timerService.schedule(Duration.ofSeconds(t.getDelaySeconds()), t.getCallback());
+      }
+    }
+
+    ctx.clearTimersAndLocks();
+  }
+
   @Override
   public void registerDelayedCallback(Duration delay, Runnable r) {
     timerService.schedule(delay, r);

src/test/java/com/uber/cadence/workflow/WorkflowTest.java

@@ -123,7 +123,7 @@ public static Object[] data() {
 
   @Rule
   public Timeout globalTimeout =
-      Timeout.seconds(DEBUGGER_TIMEOUTS ? 500 : (skipDockerService ? 5 : 20));
+      Timeout.seconds(DEBUGGER_TIMEOUTS ? 500 : (skipDockerService ? 3 : 20));
 
   @Rule
   public TestWatcher watchman =