Refactored decider to rely on initiatedEventId instead of string ids (#160)

* Refactored decider to rely on initiatedEventId instead of string ids
* Added Workflow.sideEffect
* Added Marker decision and event to TestService
* Switched TestWorkflowService to use initiatedEventId for child workflows to deal with duplicated scheduling gracefully

Author: mfateev

src/main/java/com/uber/cadence/internal/common/WorkflowExecutionUtils.java

@@ -910,4 +910,56 @@ private static void fixStackTrace(JsonElement json, String stackIndentation) {
       fixStackTrace(entry.getValue(), stackIndentation + INDENTATION);
     }
   }
+
+  /** Is this an event that was created to mirror a decision? */
+  public static boolean isDecisionEvent(HistoryEvent event) {
+    EventType eventType = event.getEventType();
+    boolean result =
+        ((event != null)
+            && (eventType == EventType.ActivityTaskScheduled
+                || eventType == EventType.StartChildWorkflowExecutionInitiated
+                || eventType == EventType.TimerStarted
+                || eventType == EventType.WorkflowExecutionCompleted
+                || eventType == EventType.WorkflowExecutionFailed
+                || eventType == EventType.WorkflowExecutionCanceled
+                || eventType == EventType.WorkflowExecutionContinuedAsNew
+                || eventType == EventType.ActivityTaskCancelRequested
+                || eventType == EventType.RequestCancelActivityTaskFailed
+                || eventType == EventType.TimerCanceled
+                || eventType == EventType.CancelTimerFailed
+                || eventType == EventType.RequestCancelExternalWorkflowExecutionInitiated
+                || eventType == EventType.MarkerRecorded
+                || eventType == EventType.SignalExternalWorkflowExecutionInitiated));
+    return result;
+  }
+
+  public static EventType getEventTypeForDecision(DecisionType decisionType) {
+    switch (decisionType) {
+      case ScheduleActivityTask:
+        return EventType.ActivityTaskScheduled;
+      case RequestCancelActivityTask:
+        return EventType.ActivityTaskCancelRequested;
+      case StartTimer:
+        return EventType.TimerStarted;
+      case CompleteWorkflowExecution:
+        return EventType.WorkflowExecutionCompleted;
+      case FailWorkflowExecution:
+        return EventType.WorkflowExecutionFailed;
+      case CancelTimer:
+        return EventType.TimerCanceled;
+      case CancelWorkflowExecution:
+        return EventType.WorkflowExecutionCanceled;
+      case RequestCancelExternalWorkflowExecution:
+        return EventType.ExternalWorkflowExecutionCancelRequested;
+      case RecordMarker:
+        return EventType.MarkerRecorded;
+      case ContinueAsNewWorkflowExecution:
+        return EventType.WorkflowExecutionContinuedAsNew;
+      case StartChildWorkflowExecution:
+        return EventType.StartChildWorkflowExecutionInitiated;
+      case SignalExternalWorkflowExecution:
+        return EventType.SignalExternalWorkflowExecutionInitiated;
+    }
+    throw new IllegalArgumentException("Unknown decisionType");
+  }
 }

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

@@ -20,7 +20,6 @@
 import com.uber.cadence.ActivityTaskCanceledEventAttributes;
 import com.uber.cadence.ActivityTaskCompletedEventAttributes;
 import com.uber.cadence.ActivityTaskFailedEventAttributes;
-import com.uber.cadence.ActivityTaskStartedEventAttributes;
 import com.uber.cadence.ActivityTaskTimedOutEventAttributes;
 import com.uber.cadence.ActivityType;
 import com.uber.cadence.HistoryEvent;
@@ -37,30 +36,35 @@ final class ActivityDecisionContext {
 
   private final class ActivityCancellationHandler implements Consumer<Exception> {
 
+    private final long scheduledEventId;
+
     private final String activityId;
 
     private final BiConsumer<byte[], Exception> callback;
 
     private ActivityCancellationHandler(
-        String activityId, BiConsumer<byte[], Exception> callaback) {
+        long scheduledEventId, String activityId, BiConsumer<byte[], Exception> callaback) {
+      this.scheduledEventId = scheduledEventId;
       this.activityId = activityId;
       this.callback = callaback;
     }
 
     @Override
     public void accept(Exception cause) {
-      if (!scheduledActivities.containsKey(activityId)) {
+      if (!scheduledActivities.containsKey(scheduledEventId)) {
         // Cancellation handlers are not deregistered. So they fire after an activity completion.
         return;
       }
       decisions.requestCancelActivityTask(
-          activityId,
+          scheduledEventId,
           () -> {
             OpenRequestInfo<byte[], ActivityType> scheduled =
-                scheduledActivities.remove(activityId);
+                scheduledActivities.remove(scheduledEventId);
             if (scheduled == null) {
               throw new IllegalArgumentException(
-                  "Activity \"" + activityId + "\" wasn't scheduled");
+                  String.format(
+                      "Activity with activityId=%s and scheduledEventId=%d wasn't found",
+                      activityId, scheduledEventId));
             }
             callback.accept(null, new CancellationException("Cancelled by request"));
           });
@@ -69,7 +73,8 @@ public void accept(Exception cause) {
 
   private final DecisionsHelper decisions;
 
-  private final Map<String, OpenRequestInfo<byte[], ActivityType>> scheduledActivities =
+  // key is scheduledEventId
+  private final Map<Long, OpenRequestInfo<byte[], ActivityType>> scheduledActivities =
       new HashMap<>();
 
   ActivityDecisionContext(DecisionsHelper decisions) {
@@ -106,21 +111,19 @@ Consumer<Exception> scheduleActivityTask(
       tl.setName(taskList);
       attributes.setTaskList(tl);
     }
-    decisions.scheduleActivityTask(attributes);
+    long scheduledEventId = decisions.scheduleActivityTask(attributes);
     context.setCompletionHandle(callback);
-    scheduledActivities.put(attributes.getActivityId(), context);
+    scheduledActivities.put(scheduledEventId, context);
     return new ActivityDecisionContext.ActivityCancellationHandler(
-        attributes.getActivityId(), callback);
+        scheduledEventId, attributes.getActivityId(), callback);
   }
 
-  void handleActivityTaskStarted(ActivityTaskStartedEventAttributes attributes) {}
-
   void handleActivityTaskCanceled(HistoryEvent event) {
     ActivityTaskCanceledEventAttributes attributes = event.getActivityTaskCanceledEventAttributes();
-    String activityId = decisions.getActivityId(attributes);
     if (decisions.handleActivityTaskCanceled(event)) {
       CancellationException e = new CancellationException();
-      OpenRequestInfo<byte[], ActivityType> scheduled = scheduledActivities.remove(activityId);
+      OpenRequestInfo<byte[], ActivityType> scheduled =
+          scheduledActivities.remove(attributes.getScheduledEventId());
       if (scheduled != null) {
         BiConsumer<byte[], Exception> completionHandle = scheduled.getCompletionCallback();
         // It is OK to fail with subclass of CancellationException when cancellation requested.
@@ -134,9 +137,9 @@ void handleActivityTaskCanceled(HistoryEvent event) {
   void handleActivityTaskCompleted(HistoryEvent event) {
     ActivityTaskCompletedEventAttributes attributes =
         event.getActivityTaskCompletedEventAttributes();
-    String activityId = decisions.getActivityId(attributes);
-    if (decisions.handleActivityTaskClosed(activityId)) {
-      OpenRequestInfo<byte[], ActivityType> scheduled = scheduledActivities.remove(activityId);
+    if (decisions.handleActivityTaskClosed(attributes.getScheduledEventId())) {
+      OpenRequestInfo<byte[], ActivityType> scheduled =
+          scheduledActivities.remove(attributes.getScheduledEventId());
       if (scheduled != null) {
         byte[] result = attributes.getResult();
         BiConsumer<byte[], Exception> completionHandle = scheduled.getCompletionCallback();
@@ -147,15 +150,15 @@ void handleActivityTaskCompleted(HistoryEvent event) {
 
   void handleActivityTaskFailed(HistoryEvent event) {
     ActivityTaskFailedEventAttributes attributes = event.getActivityTaskFailedEventAttributes();
-    String activityId = decisions.getActivityId(attributes);
-    if (decisions.handleActivityTaskClosed(activityId)) {
-      OpenRequestInfo<byte[], ActivityType> scheduled = scheduledActivities.remove(activityId);
+    if (decisions.handleActivityTaskClosed(attributes.getScheduledEventId())) {
+      OpenRequestInfo<byte[], ActivityType> scheduled =
+          scheduledActivities.remove(attributes.getScheduledEventId());
       if (scheduled != null) {
         String reason = attributes.getReason();
         byte[] details = attributes.getDetails();
         ActivityTaskFailedException failure =
             new ActivityTaskFailedException(
-                event.getEventId(), scheduled.getUserContext(), activityId, reason, details);
+                event.getEventId(), scheduled.getUserContext(), null, reason, details);
         BiConsumer<byte[], Exception> completionHandle = scheduled.getCompletionCallback();
         completionHandle.accept(null, failure);
       }
@@ -164,15 +167,15 @@ void handleActivityTaskFailed(HistoryEvent event) {
 
   void handleActivityTaskTimedOut(HistoryEvent event) {
     ActivityTaskTimedOutEventAttributes attributes = event.getActivityTaskTimedOutEventAttributes();
-    String activityId = decisions.getActivityId(attributes);
-    if (decisions.handleActivityTaskClosed(activityId)) {
-      OpenRequestInfo<byte[], ActivityType> scheduled = scheduledActivities.remove(activityId);
+    if (decisions.handleActivityTaskClosed(attributes.getScheduledEventId())) {
+      OpenRequestInfo<byte[], ActivityType> scheduled =
+          scheduledActivities.remove(attributes.getScheduledEventId());
       if (scheduled != null) {
         TimeoutType timeoutType = attributes.getTimeoutType();
         byte[] details = attributes.getDetails();
         ActivityTaskTimeoutException failure =
             new ActivityTaskTimeoutException(
-                event.getEventId(), scheduled.getUserContext(), activityId, timeoutType, details);
+                event.getEventId(), scheduled.getUserContext(), null, timeoutType, details);
         BiConsumer<byte[], Exception> completionHandle = scheduled.getCompletionCallback();
         completionHandle.accept(null, failure);
       }

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

@@ -96,15 +96,15 @@ public void handleCancellationFailureEvent(HistoryEvent event) {
   }
 
   @Override
-  public void cancel(Runnable immediateCancellationCallback) {
+  public boolean cancel(Runnable immediateCancellationCallback) {
     switch (state) {
       case STARTED:
         stateHistory.add("cancel");
         state = DecisionState.CANCELED_AFTER_STARTED;
         stateHistory.add(state.toString());
-        break;
+        return true;
       default:
-        super.cancel(immediateCancellationCallback);
+        return super.cancel(immediateCancellationCallback);
     }
   }
 

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

@@ -18,41 +18,52 @@
 package com.uber.cadence.internal.replay;
 
 import com.uber.cadence.HistoryEvent;
+import com.uber.cadence.MarkerRecordedEventAttributes;
 import com.uber.cadence.StartTimerDecisionAttributes;
 import com.uber.cadence.TimerCanceledEventAttributes;
 import com.uber.cadence.TimerFiredEventAttributes;
+import com.uber.cadence.workflow.Functions.Func;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /** Clock that must be used inside workflow definition code to ensure replay determinism. */
 final class ClockDecisionContext {
 
+  private static final String SIDE_EFFECT_MARKER_NAME = "SideEffect";
+
+  private static final Logger log = LoggerFactory.getLogger(ReplayDecider.class);
+
   private final class TimerCancellationHandler implements Consumer<Exception> {
 
-    private final String timerId;
+    private final long startEventId;
 
-    TimerCancellationHandler(String timerId) {
-      this.timerId = timerId;
+    TimerCancellationHandler(long timerId) {
+      this.startEventId = timerId;
     }
 
     @Override
     public void accept(Exception reason) {
-      decisions.cancelTimer(timerId, () -> timerCancelled(timerId, reason));
+      decisions.cancelTimer(startEventId, () -> timerCancelled(startEventId, reason));
     }
   }
 
   private final DecisionsHelper decisions;
 
-  private final Map<String, OpenRequestInfo<?, Long>> scheduledTimers = new HashMap<>();
+  // key is startedEventId
+  private final Map<Long, OpenRequestInfo<?, Long>> scheduledTimers = new HashMap<>();
 
   private long replayCurrentTimeMilliseconds;
 
   private boolean replaying = true;
 
+  private final Map<Long, byte[]> sideEffectResults = new HashMap<>();
+
   ClockDecisionContext(DecisionsHelper decisions) {
     this.decisions = decisions;
   }
@@ -81,22 +92,21 @@ Consumer<Exception> createTimer(long delaySeconds, Consumer<Exception> callback)
     final OpenRequestInfo<?, Long> context = new OpenRequestInfo<>(firingTime);
     final StartTimerDecisionAttributes timer = new StartTimerDecisionAttributes();
     timer.setStartToFireTimeoutSeconds(delaySeconds);
-    final String timerId = decisions.getNextId();
-    timer.setTimerId(timerId);
-    decisions.startTimer(timer, null);
+    timer.setTimerId(String.valueOf(decisions.getNextId()));
+    long startEventId = decisions.startTimer(timer, null);
     context.setCompletionHandle((ctx, e) -> callback.accept(e));
-    scheduledTimers.put(timerId, context);
-    return new ClockDecisionContext.TimerCancellationHandler(timerId);
+    scheduledTimers.put(startEventId, context);
+    return new ClockDecisionContext.TimerCancellationHandler(startEventId);
   }
 
   void setReplaying(boolean replaying) {
     this.replaying = replaying;
   }
 
   void handleTimerFired(TimerFiredEventAttributes attributes) {
-    String timerId = attributes.getTimerId();
-    if (decisions.handleTimerClosed(timerId)) {
-      OpenRequestInfo<?, Long> scheduled = scheduledTimers.remove(timerId);
+    long startedEventId = attributes.getStartedEventId();
+    if (decisions.handleTimerClosed(attributes)) {
+      OpenRequestInfo<?, Long> scheduled = scheduledTimers.remove(startedEventId);
       if (scheduled != null) {
         BiConsumer<?, Exception> completionCallback = scheduled.getCompletionCallback();
         completionCallback.accept(null, null);
@@ -106,14 +116,14 @@ void handleTimerFired(TimerFiredEventAttributes attributes) {
 
   void handleTimerCanceled(HistoryEvent event) {
     TimerCanceledEventAttributes attributes = event.getTimerCanceledEventAttributes();
-    String timerId = attributes.getTimerId();
+    long startedEventId = attributes.getStartedEventId();
     if (decisions.handleTimerCanceled(event)) {
-      timerCancelled(timerId, null);
+      timerCancelled(startedEventId, null);
     }
   }
 
-  private void timerCancelled(String timerId, Exception reason) {
-    OpenRequestInfo<?, ?> scheduled = scheduledTimers.remove(timerId);
+  private void timerCancelled(long startEventId, Exception reason) {
+    OpenRequestInfo<?, ?> scheduled = scheduledTimers.remove(startEventId);
     if (scheduled == null) {
       return;
     }
@@ -122,4 +132,34 @@ private void timerCancelled(String timerId, Exception reason) {
     exception.initCause(reason);
     context.accept(null, exception);
   }
+
+  byte[] sideEffect(Func<byte[]> func) {
+    long sideEffectEventId = decisions.getNextDecisionEventId();
+    byte[] result;
+    if (replaying) {
+      result = sideEffectResults.get(sideEffectEventId);
+      if (result == null) {
+        throw new Error("No cached result found for SideEffect EventID=" + sideEffectEventId);
+      }
+    } else {
+      try {
+        result = func.apply();
+      } catch (Error e) {
+        throw e;
+      } catch (Exception e) {
+        throw new Error("sideEffect function failed", e);
+      }
+    }
+    decisions.recordMarker(SIDE_EFFECT_MARKER_NAME, result);
+    return result;
+  }
+
+  void handleMarkerRecorded(HistoryEvent event) {
+    MarkerRecordedEventAttributes attributes = event.getMarkerRecordedEventAttributes();
+    if (SIDE_EFFECT_MARKER_NAME.equals(attributes.getMarkerName())) {
+      sideEffectResults.put(event.getEventId(), attributes.getDetails());
+    } else {
+      log.warn("Unexpected marker: " + event);
+    }
+  }
 }

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

@@ -46,7 +46,7 @@ public void handleInitiationFailedEvent(HistoryEvent event) {
   }
 
   @Override
-  public void cancel(Runnable immediateCancellationCallback) {
+  public boolean cancel(Runnable immediateCancellationCallback) {
     throw new UnsupportedOperationException();
   }
 

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

@@ -19,6 +19,7 @@
 
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowType;
+import com.uber.cadence.workflow.Functions.Func;
 import com.uber.cadence.workflow.Promise;
 import com.uber.m3.tally.Scope;
 import java.time.Duration;
@@ -112,6 +113,20 @@ Consumer<Exception> signalWorkflowExecution(
    */
   Consumer<Exception> createTimer(long delaySeconds, Consumer<Exception> callback);
 
+  /**
+   * Executes the provided function once, records its result into the workflow history. The recorded
+   * result on history will be returned without executing the provided function during replay. This
+   * guarantees the deterministic requirement for workflow as the exact same result will be returned
+   * in replay. Common use case is to run some short non-deterministic code in workflow, like
+   * getting random number or new UUID. The only way to fail SideEffect is to throw {@link Error}
+   * which causes decision task failure. The decision task after timeout is rescheduled and
+   * re-executed giving SideEffect another chance to succeed.
+   *
+   * @param func function that is called once to return a value.
+   * @return value of the side effect.
+   */
+  byte[] sideEffect(Func<byte[]> func);
+
   /** @return scope to be used for metrics reporting. */
   Scope getMetricsScope();