Fixing timer firing bug (#461)

* Trying to fix the timer firing bug that causes timer to not firing according to their schedule when used in Async threads.
* Add extra protection
* More fixes to the timer logic

Author: meiliang86

src/main/java/com/uber/cadence/internal/sync/DeterministicRunnerImpl.java

@@ -240,7 +240,7 @@ public void runUntilAllBlocked() throws Throwable {
         toExecuteInWorkflowThread.clear();
         progress = false;
         Iterator<WorkflowThread> ci = threads.iterator();
-        nextWakeUpTime = 0;
+        nextWakeUpTime = Long.MAX_VALUE;
         while (ci.hasNext()) {
           WorkflowThread c = ci.next();
           progress = c.runUntilBlocked() || progress;
@@ -256,7 +256,7 @@ public void runUntilAllBlocked() throws Throwable {
             }
           } else {
             long t = c.getBlockedUntil();
-            if (t > nextWakeUpTime) {
+            if (t > currentTimeMillis() && t < nextWakeUpTime) {
               nextWakeUpTime = t;
             }
           }
@@ -269,7 +269,8 @@ public void runUntilAllBlocked() throws Throwable {
           threads.addLast(c);
         }
       } while (progress && !threads.isEmpty());
-      if (nextWakeUpTime < currentTimeMillis()) {
+
+      if (nextWakeUpTime < currentTimeMillis() || nextWakeUpTime == Long.MAX_VALUE) {
         nextWakeUpTime = 0;
       }
     } finally {

src/test/java/com/uber/cadence/internal/testing/WorkflowTestingTest.java

@@ -63,6 +63,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.UUID;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -543,16 +544,17 @@ public static class TestTimerCancellationWorkflow implements TestWorkflow {
 
     @Override
     public String workflow1(String input) {
-      Promise<Void> s = Async.procedure(() -> Workflow.sleep(Duration.ofDays(1)));
+      long startTime = Workflow.currentTimeMillis();
+      Promise<Void> s = Async.procedure(() -> Workflow.sleep(Duration.ofHours(3)));
       TestActivity activity = Workflow.newActivityStub(TestActivity.class);
-      try {
-        activity.activity1("input");
-        Workflow.sleep(Duration.ofDays(3));
-      } catch (CancellationException e) {
-        return "cancelled";
-      } finally {
-        s.get();
+      activity.activity1("input");
+      Workflow.sleep(Duration.ofHours(1));
+      s.get();
+      long endTime = Workflow.currentTimeMillis();
+      if (Duration.ofMillis(endTime - startTime).compareTo(Duration.ofHours(3)) < 0) {
+        fail("workflow sleep interrupted unexpectedly");
       }
+
       return "result";
     }
   }
@@ -564,22 +566,20 @@ public void testTimerCancellation() throws TException {
     worker.registerActivitiesImplementations(new ActivityImpl());
     testEnvironment.start();
     WorkflowClient client = testEnvironment.newWorkflowClient();
-    TestWorkflow workflow = client.newWorkflowStub(TestWorkflow.class);
-    WorkflowExecution execution = WorkflowClient.start(workflow::workflow1, "input1");
-    WorkflowStub untyped = client.newUntypedWorkflowStub(execution, Optional.empty());
-    testEnvironment.sleep(Duration.ofHours(1));
-    untyped.cancel();
-    try {
-      untyped.getResult(String.class);
-      fail("unreacheable");
-    } catch (CancellationException e) {
-    }
+
+    String workflowID = UUID.randomUUID().toString();
+    TestWorkflow workflow =
+        client.newWorkflowStub(
+            TestWorkflow.class, new WorkflowOptions.Builder().setWorkflowId(workflowID).build());
+    String result = workflow.workflow1("input1");
+    assertEquals("result", result);
+
     History history =
         testEnvironment
             .getWorkflowService()
             .GetWorkflowExecutionHistory(
                 new GetWorkflowExecutionHistoryRequest()
-                    .setExecution(execution)
+                    .setExecution(new WorkflowExecution().setWorkflowId(workflowID))
                     .setDomain(client.getDomain()))
             .getHistory();
     List<HistoryEvent> historyEvents = history.getEvents();

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

@@ -432,6 +432,34 @@ public void testSync() {
         "executeActivity TestActivities::activity2");
   }
 
+  public interface TestMultipleTimers {
+    @WorkflowMethod
+    long execute();
+  }
+
+  public static class TestMultipleTimersImpl implements TestMultipleTimers {
+
+    @Override
+    public long execute() {
+      Promise<Void> t1 = Async.procedure(() -> Workflow.sleep(Duration.ofSeconds(1)));
+      Promise<Void> t2 = Async.procedure(() -> Workflow.sleep(Duration.ofSeconds(2)));
+      long start = Workflow.currentTimeMillis();
+      Promise.anyOf(t1, t2).get();
+      long elapsed = Workflow.currentTimeMillis() - start;
+      return elapsed;
+    }
+  }
+
+  @Test
+  public void testMultipleTimers() {
+    startWorkerFor(TestMultipleTimersImpl.class);
+    TestMultipleTimers workflowStub =
+        workflowClient.newWorkflowStub(
+            TestMultipleTimers.class, newWorkflowOptionsBuilder(taskList).build());
+    long result = workflowStub.execute();
+    assertTrue("should be around 1 second: " + result, result < 2000);
+  }
+
   public static class TestActivityRetryWithMaxAttempts implements TestWorkflow1 {
 
     @Override