Added Workflow.mutableSideEffect (#161)

Author: mfateev

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

@@ -17,14 +17,18 @@
 
 package com.uber.cadence.internal.replay;
 
+import com.uber.cadence.EventType;
 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.internal.replay.DecisionContext.MutableSideEffectData;
 import com.uber.cadence.workflow.Functions.Func;
+import com.uber.cadence.workflow.Functions.Func1;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
@@ -36,6 +40,7 @@
 final class ClockDecisionContext {
 
   private static final String SIDE_EFFECT_MARKER_NAME = "SideEffect";
+  private static final String MUTABLE_SIDE_EFFECT_MARKER_NAME = "MutableSideEffect";
 
   private static final Logger log = LoggerFactory.getLogger(ReplayDecider.class);
 
@@ -53,6 +58,31 @@ public void accept(Exception reason) {
     }
   }
 
+  private static final class MutableSideEffectResult {
+
+    private final byte[] data;
+
+    /**
+     * Count of how many times the mutableSideEffect was called since the last marker recorded. It
+     * is used to ensure that an updated value is returned after the same exact number of times
+     * during a replay.
+     */
+    private int accessCount;
+
+    private MutableSideEffectResult(byte[] data) {
+      this.data = data;
+    }
+
+    public byte[] getData() {
+      accessCount++;
+      return data;
+    }
+
+    public int getAccessCount() {
+      return accessCount;
+    }
+  }
+
   private final DecisionsHelper decisions;
 
   // key is startedEventId
@@ -62,7 +92,10 @@ public void accept(Exception reason) {
 
   private boolean replaying = true;
 
+  // Key is side effect marker eventId
   private final Map<Long, byte[]> sideEffectResults = new HashMap<>();
+  // Key is mutableSideEffect id
+  private final Map<String, MutableSideEffectResult> mutableSideEffectResults = new HashMap<>();
 
   ClockDecisionContext(DecisionsHelper decisions) {
     this.decisions = decisions;
@@ -96,7 +129,7 @@ Consumer<Exception> createTimer(long delaySeconds, Consumer<Exception> callback)
     long startEventId = decisions.startTimer(timer, null);
     context.setCompletionHandle((ctx, e) -> callback.accept(e));
     scheduledTimers.put(startEventId, context);
-    return new ClockDecisionContext.TimerCancellationHandler(startEventId);
+    return new TimerCancellationHandler(startEventId);
   }
 
   void setReplaying(boolean replaying) {
@@ -154,11 +187,87 @@ byte[] sideEffect(Func<byte[]> func) {
     return result;
   }
 
+  /**
+   * @param id mutable side effect id
+   * @param func given the value from the last marker returns value to store. If result is empty
+   *     nothing is recorded into the history.
+   * @return the latest value returned by func
+   */
+  Optional<byte[]> mutableSideEffect(
+      String id,
+      Func1<MutableSideEffectData, byte[]> markerDataSerializer,
+      Func1<byte[], MutableSideEffectData> markerDataDeserializer,
+      Func1<Optional<byte[]>, Optional<byte[]>> func) {
+    MutableSideEffectResult result = mutableSideEffectResults.get(id);
+    Optional<byte[]> stored;
+    if (result == null) {
+      stored = Optional.empty();
+    } else {
+      stored = Optional.of(result.getData());
+    }
+    long eventId = decisions.getNextDecisionEventId();
+    int accessCount = result == null ? 0 : result.getAccessCount();
+    if (replaying) {
+
+      Optional<byte[]> data =
+          getSideEffectDataFromHistory(eventId, id, accessCount, markerDataDeserializer);
+      if (data.isPresent()) {
+        // Need to insert marker to ensure that eventId is incremented
+        recordMutableSideEffectMarker(id, eventId, data.get(), accessCount, markerDataSerializer);
+        return data;
+      }
+      return stored;
+    }
+    Optional<byte[]> toStore = func.apply(stored);
+    if (toStore.isPresent()) {
+      byte[] data = toStore.get();
+      recordMutableSideEffectMarker(id, eventId, data, accessCount, markerDataSerializer);
+      return toStore;
+    }
+    return stored;
+  }
+
+  private Optional<byte[]> getSideEffectDataFromHistory(
+      long eventId,
+      String mutableSideEffectId,
+      int expectedAcccessCount,
+      Func1<byte[], MutableSideEffectData> markerDataDeserializer) {
+    HistoryEvent event = decisions.getDecisionEvent(eventId);
+    if (event.getEventType() != EventType.MarkerRecorded) {
+      return Optional.empty();
+    }
+    MarkerRecordedEventAttributes attributes = event.getMarkerRecordedEventAttributes();
+    String name = attributes.getMarkerName();
+    if (!MUTABLE_SIDE_EFFECT_MARKER_NAME.equals(name)) {
+      return Optional.empty();
+    }
+    MutableSideEffectData markerData = markerDataDeserializer.apply(attributes.getDetails());
+    // access count is used to not return data from the marker before the recorded number of calls
+    if (!mutableSideEffectId.equals(markerData.getId())
+        || markerData.getAccessCount() > expectedAcccessCount) {
+      return Optional.empty();
+    }
+    return Optional.of(markerData.getData());
+  }
+
+  private void recordMutableSideEffectMarker(
+      String id,
+      long eventId,
+      byte[] data,
+      int accessCount,
+      Func1<MutableSideEffectData, byte[]> markerDataSerializer) {
+    MutableSideEffectData dataObject = new MutableSideEffectData(id, eventId, data, accessCount);
+    byte[] details = markerDataSerializer.apply(dataObject);
+    mutableSideEffectResults.put(id, new MutableSideEffectResult(data));
+    decisions.recordMarker(MUTABLE_SIDE_EFFECT_MARKER_NAME, details);
+  }
+
   void handleMarkerRecorded(HistoryEvent event) {
     MarkerRecordedEventAttributes attributes = event.getMarkerRecordedEventAttributes();
-    if (SIDE_EFFECT_MARKER_NAME.equals(attributes.getMarkerName())) {
+    String name = attributes.getMarkerName();
+    if (SIDE_EFFECT_MARKER_NAME.equals(name)) {
       sideEffectResults.put(event.getEventId(), attributes.getDetails());
-    } else {
+    } else if (!MUTABLE_SIDE_EFFECT_MARKER_NAME.equals(name)) {
       log.warn("Unexpected marker: " + event);
     }
   }

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

@@ -20,9 +20,11 @@
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowType;
 import com.uber.cadence.workflow.Functions.Func;
+import com.uber.cadence.workflow.Functions.Func1;
 import com.uber.cadence.workflow.Promise;
 import com.uber.m3.tally.Scope;
 import java.time.Duration;
+import java.util.Optional;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
@@ -32,6 +34,37 @@
  */
 public interface DecisionContext extends ReplayAware {
 
+  final class MutableSideEffectData {
+
+    private final String id;
+    private final long eventId;
+    private final byte[] data;
+    private final int accessCount;
+
+    public MutableSideEffectData(String id, long eventId, byte[] data, int accessCount) {
+      this.id = id;
+      this.eventId = eventId;
+      this.data = data;
+      this.accessCount = accessCount;
+    }
+
+    public String getId() {
+      return id;
+    }
+
+    public long getEventId() {
+      return eventId;
+    }
+
+    public byte[] getData() {
+      return data;
+    }
+
+    public int getAccessCount() {
+      return accessCount;
+    }
+  }
+
   WorkflowExecution getWorkflowExecution();
 
   // TODO: Add to Cadence
@@ -97,6 +130,12 @@ Consumer<Exception> signalWorkflowExecution(
   /** Deterministic unique Id generator */
   String generateUniqueId();
 
+  Optional<byte[]> mutableSideEffect(
+      String id,
+      Func1<MutableSideEffectData, byte[]> markerDataSerializer,
+      Func1<byte[], MutableSideEffectData> markerDataDeserializer,
+      Func1<Optional<byte[]>, Optional<byte[]>> func);
+
   /**
    * @return time of the {@link com.uber.cadence.PollForDecisionTaskResponse} start event of the
    *     decision being processed or replayed.

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

@@ -25,10 +25,12 @@
 import com.uber.cadence.WorkflowType;
 import com.uber.cadence.internal.metrics.ReplayAwareScope;
 import com.uber.cadence.workflow.Functions.Func;
+import com.uber.cadence.workflow.Functions.Func1;
 import com.uber.cadence.workflow.Promise;
 import com.uber.cadence.workflow.Workflow;
 import com.uber.m3.tally.Scope;
 import java.time.Duration;
+import java.util.Optional;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
@@ -197,6 +199,15 @@ public byte[] sideEffect(Func<byte[]> func) {
     return workflowClock.sideEffect(func);
   }
 
+  @Override
+  public Optional<byte[]> mutableSideEffect(
+      String id,
+      Func1<MutableSideEffectData, byte[]> markerDataSerializer,
+      Func1<byte[], MutableSideEffectData> markerDataDeserializer,
+      Func1<Optional<byte[]>, Optional<byte[]>> func) {
+    return workflowClock.mutableSideEffect(id, markerDataSerializer, markerDataDeserializer, func);
+  }
+
   @Override
   public long currentTimeMillis() {
     return workflowClock.currentTimeMillis();

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

@@ -51,6 +51,7 @@
 import com.uber.cadence.WorkflowExecutionStartedEventAttributes;
 import com.uber.cadence.WorkflowType;
 import com.uber.cadence.internal.common.WorkflowExecutionUtils;
+import com.uber.cadence.internal.replay.HistoryHelper.DecisionEvents;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -86,6 +87,8 @@ class DecisionsHelper {
    */
   private long nextDecisionEventId;
 
+  private DecisionEvents decisionEvents;
+
   /** Use access-order to ensure that decisions are emitted in order of their creation */
   private final Map<DecisionId, DecisionStateMachine> decisions =
       new LinkedHashMap<>(100, 0.75f, true);
@@ -514,8 +517,9 @@ private boolean isCompletionEvent(Decision decision) {
     }
   }
 
-  public void handleDecisionTaskStartedEvent(long nextDecisionEventId) {
-    this.nextDecisionEventId = nextDecisionEventId;
+  public void handleDecisionTaskStartedEvent(DecisionEvents decision) {
+    this.decisionEvents = decision;
+    this.nextDecisionEventId = decision.getNextDecisionEventId();
   }
 
   void notifyDecisionSent() {
@@ -599,4 +603,8 @@ String getNextId() {
     }
     return String.valueOf(nextDecisionEventId);
   }
+
+  public HistoryEvent getDecisionEvent(long eventId) {
+    return decisionEvents.getDecisionEvent(eventId);
+  }
 }

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

@@ -82,6 +82,14 @@ public List<HistoryEvent> getDecisionEvents() {
       return decisionEvents;
     }
 
+    public HistoryEvent getDecisionEvent(long eventId) {
+      int index = (int) (eventId - nextDecisionEventId);
+      if (index < 0 || index > decisionEvents.size()) {
+        throw new IllegalArgumentException("No decision event found at eventId=" + eventId);
+      }
+      return decisionEvents.get(index);
+    }
+
     public List<HistoryEvent> getMarkers() {
       return markers;
     }

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

@@ -363,7 +363,7 @@ private void decideImpl(Functions.Proc query) throws Throwable {
         context.setReplaying(decision.isReplay());
         context.setReplayCurrentTimeMilliseconds(decision.getReplayCurrentTimeMilliseconds());
 
-        decisionsHelper.handleDecisionTaskStartedEvent(decision.getNextDecisionEventId());
+        decisionsHelper.handleDecisionTaskStartedEvent(decision);
         // Markers must be cached first as their data is needed when processing events.
         for (HistoryEvent event : decision.getMarkers()) {
           processEvent(event);

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

@@ -27,6 +27,7 @@
 import com.uber.cadence.internal.replay.SignalExternalWorkflowParameters;
 import com.uber.cadence.internal.replay.StartChildWorkflowExecutionParameters;
 import com.uber.cadence.workflow.Functions.Func;
+import com.uber.cadence.workflow.Functions.Func1;
 import com.uber.cadence.workflow.Promise;
 import com.uber.m3.tally.Scope;
 import java.time.Duration;
@@ -504,6 +505,15 @@ public String generateUniqueId() {
       throw new UnsupportedOperationException("not implemented");
     }
 
+    @Override
+    public Optional<byte[]> mutableSideEffect(
+        String id,
+        Func1<MutableSideEffectData, byte[]> markerDataSerializer,
+        Func1<byte[], MutableSideEffectData> markerDataDeserializer,
+        Func1<Optional<byte[]>, Optional<byte[]>> func) {
+      throw new UnsupportedOperationException("not implemented");
+    }
+
     @Override
     public long currentTimeMillis() {
       throw new UnsupportedOperationException("not implemented");