temporalio / sdk-java / #172

/*

 * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.

 *

 * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.

 *

 * Modifications copyright (C) 2017 Uber Technologies, Inc.

 *

 * Licensed under the Apache License, Version 2.0 (the "License");

 * you may not use this material except in compliance with the License.

 * You may obtain a copy of the License at

 *

 *   http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package io.temporal.internal.testservice;

import static io.temporal.internal.testservice.StateMachines.Action.CANCEL;

import static io.temporal.internal.testservice.StateMachines.Action.COMPLETE;

import static io.temporal.internal.testservice.StateMachines.Action.CONTINUE_AS_NEW;

import static io.temporal.internal.testservice.StateMachines.Action.FAIL;

import static io.temporal.internal.testservice.StateMachines.Action.INITIATE;

import static io.temporal.internal.testservice.StateMachines.Action.QUERY;

import static io.temporal.internal.testservice.StateMachines.Action.REQUEST_CANCELLATION;

import static io.temporal.internal.testservice.StateMachines.Action.START;

import static io.temporal.internal.testservice.StateMachines.Action.TERMINATE;

import static io.temporal.internal.testservice.StateMachines.Action.TIME_OUT;

import static io.temporal.internal.testservice.StateMachines.Action.UPDATE;

import static io.temporal.internal.testservice.StateMachines.Action.UPDATE_WORKFLOW_EXECUTION;

import static io.temporal.internal.testservice.StateMachines.State.CANCELED;

import static io.temporal.internal.testservice.StateMachines.State.CANCELLATION_REQUESTED;

import static io.temporal.internal.testservice.StateMachines.State.COMPLETED;

import static io.temporal.internal.testservice.StateMachines.State.CONTINUED_AS_NEW;

import static io.temporal.internal.testservice.StateMachines.State.FAILED;

import static io.temporal.internal.testservice.StateMachines.State.INITIATED;

import static io.temporal.internal.testservice.StateMachines.State.NONE;

import static io.temporal.internal.testservice.StateMachines.State.STARTED;

import static io.temporal.internal.testservice.StateMachines.State.TERMINATED;

import static io.temporal.internal.testservice.StateMachines.State.TIMED_OUT;

import com.google.common.base.Preconditions;

import com.google.protobuf.Any;

import com.google.protobuf.Duration;

import com.google.protobuf.InvalidProtocolBufferException;

import com.google.protobuf.Timestamp;

import com.google.protobuf.util.Durations;

import com.google.protobuf.util.Timestamps;

import io.grpc.Status;

import io.grpc.StatusRuntimeException;

import io.temporal.api.command.v1.CancelTimerCommandAttributes;

import io.temporal.api.command.v1.CancelWorkflowExecutionCommandAttributes;

import io.temporal.api.command.v1.CompleteWorkflowExecutionCommandAttributes;

import io.temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes;

import io.temporal.api.command.v1.FailWorkflowExecutionCommandAttributes;

import io.temporal.api.command.v1.RequestCancelActivityTaskCommandAttributes;

import io.temporal.api.command.v1.RequestCancelExternalWorkflowExecutionCommandAttributes;

import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributes;

import io.temporal.api.command.v1.SignalExternalWorkflowExecutionCommandAttributes;

import io.temporal.api.command.v1.StartChildWorkflowExecutionCommandAttributes;

import io.temporal.api.command.v1.StartTimerCommandAttributes;

import io.temporal.api.common.v1.Payloads;

import io.temporal.api.common.v1.RetryPolicy;

import io.temporal.api.common.v1.WorkflowExecution;

import io.temporal.api.enums.v1.CancelExternalWorkflowExecutionFailedCause;

import io.temporal.api.enums.v1.EventType;

import io.temporal.api.enums.v1.RetryState;

import io.temporal.api.enums.v1.SignalExternalWorkflowExecutionFailedCause;

import io.temporal.api.enums.v1.StartChildWorkflowExecutionFailedCause;

import io.temporal.api.enums.v1.TimeoutType;

import io.temporal.api.errordetails.v1.QueryFailedFailure;

import io.temporal.api.failure.v1.ApplicationFailureInfo;

import io.temporal.api.failure.v1.Failure;

import io.temporal.api.failure.v1.TimeoutFailureInfo;

import io.temporal.api.history.v1.*;

import io.temporal.api.protocol.v1.Message;

import io.temporal.api.query.v1.WorkflowQueryResult;

import io.temporal.api.taskqueue.v1.StickyExecutionAttributes;

import io.temporal.api.taskqueue.v1.TaskQueue;

import io.temporal.api.update.v1.*;

import io.temporal.api.workflowservice.v1.*;

import io.temporal.internal.common.ProtobufTimeUtils;

import io.temporal.internal.testservice.TestWorkflowMutableStateImpl.UpdateWorkflowExecution;

import io.temporal.internal.testservice.TestWorkflowStore.ActivityTask;

import io.temporal.internal.testservice.TestWorkflowStore.TaskQueueId;

import io.temporal.internal.testservice.TestWorkflowStore.WorkflowTask;

import io.temporal.serviceclient.StatusUtils;

import io.temporal.workflow.Functions;

import java.util.*;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.ForkJoinPool;

import javax.annotation.Nonnull;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

class StateMachines {

  private static final Logger log = LoggerFactory.getLogger(StateMachines.class);

  public static final long DEFAULT_WORKFLOW_EXECUTION_TIMEOUT_MILLISECONDS =

      10L * 365 * 24 * 3600 * 1000;

  public static final long DEFAULT_WORKFLOW_TASK_TIMEOUT_MILLISECONDS = 10L * 1000;

  public static final long MAX_WORKFLOW_TASK_TIMEOUT_MILLISECONDS = 60L * 1000;

  static final Duration DEFAULT_ACTIVITY_RETRY_INITIAL_INTERVAL = Durations.fromSeconds(1);

  static final double DEFAULT_ACTIVITY_RETRY_BACKOFF_COEFFICIENT = 2.0;

  static final int DEFAULT_ACTIVITY_RETRY_MAXIMUM_ATTEMPTS = 0;

  static final int DEFAULT_ACTIVITY_MAXIMUM_INTERVAL_COEFFICIENT = 100;

  static final int NO_EVENT_ID = -1;

  enum State {

    NONE,

    INITIATED,

    STARTED,

    FAILED,

    TIMED_OUT,

    CANCELLATION_REQUESTED,

    CANCELED,

    COMPLETED,

    CONTINUED_AS_NEW,

    TERMINATED,

  }

  enum Action {

    INITIATE,

    START,

    FAIL,

    TIME_OUT,

    REQUEST_CANCELLATION,

    CANCEL,

    TERMINATE,

    UPDATE,

    COMPLETE,

    CONTINUE_AS_NEW,

    QUERY,

    UPDATE_WORKFLOW_EXECUTION,

  }

  static final class WorkflowData {

    Optional<TestServiceRetryState> retryState;

    Duration backoffStartInterval;

    String cronSchedule;

    Payloads lastCompletionResult;

    Optional<Failure> lastFailure;

    /**

     * @see WorkflowExecutionStartedEventAttributes#getFirstExecutionRunId()

     */

    final @Nonnull String firstExecutionRunId;

    /**

     * @see WorkflowExecutionStartedEventAttributes#getOriginalExecutionRunId()

     */

    final @Nonnull String originalExecutionRunId;

    /** RunId that was continued by this run as a result of Retry or Continue-As-New. */

    Optional<String> continuedExecutionRunId;

    Functions.Proc runTimerCancellationHandle;

    WorkflowData(

        Optional<TestServiceRetryState> retryState,

        Duration backoffStartInterval,

        String cronSchedule,

        Payloads lastCompletionResult,

        Optional<Failure> lastFailure,

        @Nonnull String firstExecutionRunId,

        @Nonnull String originalExecutionRunId,

        Optional<String> continuedExecutionRunId) {

      this.retryState = retryState;

      this.backoffStartInterval = backoffStartInterval;

      this.cronSchedule = cronSchedule;

      this.lastCompletionResult = lastCompletionResult;

      this.firstExecutionRunId =

          Preconditions.checkNotNull(firstExecutionRunId, "firstExecutionRunId");

      this.originalExecutionRunId =

          Preconditions.checkNotNull(originalExecutionRunId, "originalExecutionRunId");

      this.continuedExecutionRunId = continuedExecutionRunId;

      this.lastFailure = Objects.requireNonNull(lastFailure);

    }

    @Override

    public String toString() {

      return "WorkflowData{"

          + "retryState="

          + retryState

          + ", backoffStartInterval="

          + backoffStartInterval

          + ", cronSchedule='"

          + cronSchedule

          + '\''

          + ", lastCompletionResult="

          + lastCompletionResult

          + ", firstExecutionRunId='"

          + firstExecutionRunId

          + '\''

          + ", originalExecutionRunId='"

          + originalExecutionRunId

          + '\''

          + ", continuedExecutionRunId="

          + continuedExecutionRunId

          + '}';

    }

  }

  static final class WorkflowTaskData {

    final TestWorkflowStore store;

    boolean workflowCompleted;

    /** id of the last started event which completed successfully */

    long lastSuccessfulStartedEventId;

    final StartWorkflowExecutionRequest startRequest;

    long startedEventId = NO_EVENT_ID;

    PollWorkflowTaskQueueResponse.Builder workflowTask;

    /**

     * Events that are added during execution of a workflow task. They have to be buffered to be

     * added after the events generated by a workflow task. Without this the determinism will be

     * broken on replay.

     */

    final List<RequestContext> bufferedEvents = new ArrayList<>();

    /**

     * Update requests that are added during execution of a workflow task. They have to be buffered

     * to be added to the next workflow task.

     */

    final Map<String, UpdateWorkflowExecution> updateRequestBuffer = new LinkedHashMap<>();

    final Map<String, UpdateWorkflowExecution> updateRequest = new LinkedHashMap<>();

    long scheduledEventId = NO_EVENT_ID;

    int attempt = 0;

    /** Query requests received during workflow task processing (after start) */

    final Map<String, TestWorkflowMutableStateImpl.ConsistentQuery> queryBuffer = new HashMap<>();

    final Map<String, TestWorkflowMutableStateImpl.ConsistentQuery> consistentQueryRequests =

        new HashMap<>();

    WorkflowTaskData(TestWorkflowStore store, StartWorkflowExecutionRequest startRequest) {

      this.store = store;

      this.startRequest = startRequest;

    }

    void clear() {

      startedEventId = NO_EVENT_ID;

      workflowTask = null;

      scheduledEventId = NO_EVENT_ID;

      attempt = 0;

    }

    Optional<UpdateWorkflowExecution> getUpdateRequest(String protocolInstanceId) {

      return Optional.ofNullable(

          updateRequest.getOrDefault(

              protocolInstanceId, updateRequestBuffer.get(protocolInstanceId)));

    }

    @Override

    public String toString() {

      return "WorkflowTaskData{"

          + "store="

          + store

          + ", workflowCompleted="

          + workflowCompleted

          + ", lastSuccessfulStartedEventId="

          + lastSuccessfulStartedEventId

          + ", startRequest="

          + startRequest

          + ", startedEventId="

          + startedEventId

          + ", workflowTask="

          + workflowTask

          + ", bufferedEvents="

          + bufferedEvents

          + ", scheduledEventId="

          + scheduledEventId

          + ", attempt="

          + attempt

          + ", queryBuffer="

          + queryBuffer

          + ", consistentQueryRequests="

          + consistentQueryRequests

          + ", updateRequest="

          + updateRequest

          + ", updateRequestBuffer="

          + updateRequestBuffer

          + '}';

    }

  }

  static final class ActivityTaskData {

    StartWorkflowExecutionRequest startWorkflowExecutionRequest;

    ActivityTaskScheduledEventAttributes scheduledEvent;

    ActivityTask activityTask;

    final TestWorkflowStore store;

    long scheduledEventId = NO_EVENT_ID;

    long startedEventId = NO_EVENT_ID;

    public HistoryEvent startedEvent;

    Payloads heartbeatDetails;

    long lastHeartbeatTime;

    TestServiceRetryState retryState;

    Duration nextBackoffInterval;

    String identity;

    ActivityTaskData(

        TestWorkflowStore store, StartWorkflowExecutionRequest startWorkflowExecutionRequest) {

      this.store = store;

      this.startWorkflowExecutionRequest = startWorkflowExecutionRequest;

    }

    @Override

    public String toString() {

      return "ActivityTaskData{"

          + "startWorkflowExecutionRequest="

          + startWorkflowExecutionRequest

          + ", scheduledEvent="

          + scheduledEvent

          + ", activityTask="

          + activityTask

          + ", store="

          + store

          + ", scheduledEventId="

          + scheduledEventId

          + ", startedEventId="

          + startedEventId

          + ", startedEvent="

          + startedEvent

          + ", heartbeatDetails="

          + heartbeatDetails

          + ", lastHeartbeatTime="

          + lastHeartbeatTime

          + ", retryState="

          + retryState

          + ", nextBackoffInterval="

          + nextBackoffInterval

          + '}';

    }

    public int getAttempt() {

      return retryState != null ? retryState.getAttempt() : 1;

    }

  }

  static final class SignalExternalData {

    long initiatedEventId = NO_EVENT_ID;

    public SignalExternalWorkflowExecutionInitiatedEventAttributes initiatedEvent;

    @Override

    public String toString() {

      return "SignalExternalData{"

          + "initiatedEventId="

          + initiatedEventId

          + ", initiatedEvent="

          + initiatedEvent

          + '}';

    }

  }

  static final class CancelExternalData {

    long initiatedEventId = NO_EVENT_ID;

    public RequestCancelExternalWorkflowExecutionInitiatedEventAttributes initiatedEvent;

    @Override

    public String toString() {

      return "CancelExternalData{"

          + "initiatedEventId="

          + initiatedEventId

          + ", initiatedEvent="

          + initiatedEvent

          + '}';

    }

  }

  static final class ChildWorkflowData {

    final TestWorkflowService service;

    StartChildWorkflowExecutionInitiatedEventAttributes initiatedEvent;

    long initiatedEventId;

    long startedEventId;

    WorkflowExecution execution;

    public ChildWorkflowData(TestWorkflowService service) {

      this.service = service;

    }

    @Override

    public String toString() {

      return "ChildWorkflowData{"

          + "service="

          + service

          + ", initiatedEvent="

          + initiatedEvent

          + ", initiatedEventId="

          + initiatedEventId

          + ", startedEventId="

          + startedEventId

          + ", execution="

          + execution

          + '}';

    }

  }

  static final class TimerData {

    TimerStartedEventAttributes startedEvent;

    public long startedEventId;

    @Override

    public String toString() {

      return "TimerData{"

          + "startedEvent="

          + startedEvent

          + ", startedEventId="

          + startedEventId

          + '}';

    }

  }

  /** Represents an accepted update workflow execution request */

  static final class UpdateWorkflowExecutionData {

    final String id;

    final CompletableFuture<UpdateWorkflowExecutionResponse> acceptance;

    final CompletableFuture<UpdateWorkflowExecutionResponse> complete;

    public UpdateWorkflowExecutionData(

        String id,

        CompletableFuture<UpdateWorkflowExecutionResponse> acceptance,

        CompletableFuture<UpdateWorkflowExecutionResponse> complete) {

      this.id = id;

      this.acceptance = acceptance;

      this.complete = complete;

    }

    @Override

    public String toString() {

      return "UpdateWorkflowExecutionData{" + "ID=" + id + '}';

    }

  }

  static StateMachine<WorkflowData> newWorkflowStateMachine(WorkflowData data) {

    return new StateMachine<>(data)

        .add(NONE, START, STARTED, StateMachines::startWorkflow)

        .add(STARTED, COMPLETE, COMPLETED, StateMachines::completeWorkflow)

        .add(STARTED, CONTINUE_AS_NEW, CONTINUED_AS_NEW, StateMachines::continueAsNewWorkflow)

        .add(STARTED, FAIL, FAILED, StateMachines::failWorkflow)

        .add(STARTED, TIME_OUT, TIMED_OUT, StateMachines::timeoutWorkflow)

        .add(

            STARTED,

            REQUEST_CANCELLATION,

            CANCELLATION_REQUESTED,

            StateMachines::requestWorkflowCancellation)

        .add(STARTED, TERMINATE, TERMINATED, StateMachines::terminateWorkflow)

        .add(

            CANCELLATION_REQUESTED,

            REQUEST_CANCELLATION,

            CANCELLATION_REQUESTED,

            StateMachines::noop)

        .add(CANCELLATION_REQUESTED, COMPLETE, COMPLETED, StateMachines::completeWorkflow)

        .add(CANCELLATION_REQUESTED, CANCEL, CANCELED, StateMachines::cancelWorkflow)

        .add(CANCELLATION_REQUESTED, TERMINATE, TERMINATED, StateMachines::terminateWorkflow)

        .add(CANCELLATION_REQUESTED, FAIL, FAILED, StateMachines::failWorkflow)

        .add(CANCELLATION_REQUESTED, TIME_OUT, TIMED_OUT, StateMachines::timeoutWorkflow);

  }

  static StateMachine<WorkflowTaskData> newWorkflowTaskStateMachine(

      TestWorkflowStore store, StartWorkflowExecutionRequest startRequest) {

    return new StateMachine<>(new WorkflowTaskData(store, startRequest))

        .add(NONE, INITIATE, INITIATED, StateMachines::scheduleWorkflowTask)

        // TODO(maxim): Uncomment once the server supports consistent query only workflow tasks

        //        .add(NONE, QUERY, INITIATED_QUERY_ONLY, StateMachines::scheduleQueryWorkflowTask)

        //        .add(INITIATED_QUERY_ONLY, QUERY, INITIATED_QUERY_ONLY,

        // StateMachines::queryWhileScheduled)

        //        .add(

        //            INITIATED_QUERY_ONLY,

        //            INITIATE,

        //            INITIATED,

        //            StateMachines::convertQueryWorkflowTaskToReal)

        //        .add(

        //            INITIATED_QUERY_ONLY,

        //            START,

        //            STARTED_QUERY_ONLY,

        //            StateMachines::startQueryOnlyWorkflowTask)

        //        .add(STARTED_QUERY_ONLY, INITIATE, STARTED_QUERY_ONLY,

        // StateMachines::needsWorkflowTask)

        //        .add(STARTED_QUERY_ONLY, QUERY, STARTED_QUERY_ONLY,

        // StateMachines::needsWorkflowTaskDueToQuery)

        //        .add(STARTED_QUERY_ONLY, FAIL, NONE, StateMachines::failQueryWorkflowTask)

        //        .add(STARTED_QUERY_ONLY, TIME_OUT, NONE, StateMachines::failQueryWorkflowTask)

        //        .add(STARTED_QUERY_ONLY, COMPLETE, NONE, StateMachines::completeQuery)

        .add(STARTED, QUERY, STARTED, StateMachines::bufferQuery)

        .add(STARTED, UPDATE_WORKFLOW_EXECUTION, STARTED, StateMachines::bufferUpdate)

        .add(INITIATED, INITIATE, INITIATED, StateMachines::noop)

        .add(INITIATED, QUERY, INITIATED, StateMachines::queryWhileScheduled)

        .add(INITIATED, UPDATE_WORKFLOW_EXECUTION, INITIATED, StateMachines::addUpdate)

        .add(INITIATED, START, STARTED, StateMachines::startWorkflowTask)

        .add(STARTED, COMPLETE, NONE, StateMachines::completeWorkflowTask)

        .add(STARTED, FAIL, NONE, StateMachines::failWorkflowTask)

        .add(STARTED, TIME_OUT, NONE, StateMachines::timeoutWorkflowTask)

        .add(STARTED, INITIATE, STARTED, StateMachines::needsWorkflowTask);

  }

  public static StateMachine<ActivityTaskData> newActivityStateMachine(

      TestWorkflowStore store, StartWorkflowExecutionRequest workflowStartedEvent) {

    return new StateMachine<>(new ActivityTaskData(store, workflowStartedEvent))

        .add(NONE, INITIATE, INITIATED, StateMachines::scheduleActivityTask)

        .add(INITIATED, START, STARTED, StateMachines::startActivityTask)

        .add(INITIATED, TIME_OUT, TIMED_OUT, StateMachines::timeoutActivityTask)

        .add(

            INITIATED,

            REQUEST_CANCELLATION,

            CANCELLATION_REQUESTED,

            StateMachines::requestActivityCancellation)

        .add(STARTED, COMPLETE, COMPLETED, StateMachines::completeActivityTask)

        // Transitions to initiated in case of a retry

        .add(STARTED, FAIL, new State[] {FAILED, INITIATED}, StateMachines::failActivityTask)

        // Transitions to initiated in case of a retry

        .add(

            STARTED,

            TIME_OUT,

            new State[] {TIMED_OUT, INITIATED},

            StateMachines::timeoutActivityTask)

        .add(STARTED, UPDATE, STARTED, StateMachines::heartbeatActivityTask)

        .add(

            STARTED,

            REQUEST_CANCELLATION,

            CANCELLATION_REQUESTED,

            StateMachines::requestActivityCancellation)

        .add(

            CANCELLATION_REQUESTED, CANCEL, CANCELED, StateMachines::reportActivityTaskCancellation)

        .add(CANCELLATION_REQUESTED, COMPLETE, COMPLETED, StateMachines::completeActivityTask)

        .add(

            CANCELLATION_REQUESTED,

            UPDATE,

            CANCELLATION_REQUESTED,

            StateMachines::heartbeatActivityTask)

        .add(CANCELLATION_REQUESTED, TIME_OUT, TIMED_OUT, StateMachines::timeoutActivityTask)

        .add(CANCELLATION_REQUESTED, FAIL, FAILED, StateMachines::failActivityTask);

  }

  public static StateMachine<ChildWorkflowData> newChildWorkflowStateMachine(

      TestWorkflowService service) {

    return new StateMachine<>(new ChildWorkflowData(service))

        .add(NONE, INITIATE, INITIATED, StateMachines::initiateChildWorkflow)

        .add(INITIATED, START, STARTED, StateMachines::childWorkflowStarted)

        .add(INITIATED, FAIL, FAILED, StateMachines::startChildWorkflowFailed)

        .add(INITIATED, TIME_OUT, TIMED_OUT, StateMachines::timeoutChildWorkflow)

        .add(STARTED, COMPLETE, COMPLETED, StateMachines::childWorkflowCompleted)

        .add(STARTED, FAIL, FAILED, StateMachines::childWorkflowFailed)

        .add(STARTED, TIME_OUT, TIMED_OUT, StateMachines::timeoutChildWorkflow)

        .add(STARTED, CANCEL, CANCELED, StateMachines::childWorkflowCanceled);

  }

  public static StateMachine<UpdateWorkflowExecutionData> newUpdateWorkflowExecution(

      String updateId,

      CompletableFuture<UpdateWorkflowExecutionResponse> acceptance,

      CompletableFuture<UpdateWorkflowExecutionResponse> complete) {

    return new StateMachine<>(new UpdateWorkflowExecutionData(updateId, acceptance, complete))

        .add(NONE, START, STARTED, StateMachines::acceptUpdate)

        .add(STARTED, COMPLETE, COMPLETED, StateMachines::completeUpdate);

  }

  public static StateMachine<TimerData> newTimerStateMachine() {

    return new StateMachine<>(new TimerData())

        .add(NONE, START, STARTED, StateMachines::startTimer)

        .add(STARTED, COMPLETE, COMPLETED, StateMachines::fireTimer)

        .add(STARTED, CANCEL, CANCELED, StateMachines::cancelTimer);

  }

  public static StateMachine<SignalExternalData> newSignalExternalStateMachine() {

    return new StateMachine<>(new SignalExternalData())

        .add(NONE, INITIATE, INITIATED, StateMachines::initiateExternalSignal)

        .add(INITIATED, FAIL, FAILED, StateMachines::failExternalSignal)

        .add(INITIATED, COMPLETE, COMPLETED, StateMachines::completeExternalSignal);

  }

  public static StateMachine<CancelExternalData> newCancelExternalStateMachine() {

    return new StateMachine<>(new CancelExternalData())

        .add(NONE, INITIATE, INITIATED, StateMachines::initiateExternalCancellation)

        .add(INITIATED, FAIL, FAILED, StateMachines::failExternalCancellation)

        .add(INITIATED, START, STARTED, StateMachines::reportExternalCancellationRequested);

  }

  private static <T, A> void noop(RequestContext ctx, T data, A a, long notUsed) {}

  private static void timeoutChildWorkflow(

      RequestContext ctx, ChildWorkflowData data, RetryState retryState, long notUsed) {

    StartChildWorkflowExecutionInitiatedEventAttributes ie = data.initiatedEvent;

    ChildWorkflowExecutionTimedOutEventAttributes a =

        ChildWorkflowExecutionTimedOutEventAttributes.newBuilder()

            .setNamespace(ie.getNamespace())

            .setStartedEventId(data.startedEventId)

            .setWorkflowExecution(data.execution)

            .setWorkflowType(ie.getWorkflowType())

            .setRetryState(retryState)

            .setInitiatedEventId(data.initiatedEventId)

            .build();

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_TIMED_OUT)

            .setChildWorkflowExecutionTimedOutEventAttributes(a)

            .build();

    ctx.addEvent(event);

  }

  private static void startChildWorkflowFailed(

      RequestContext ctx,

      ChildWorkflowData data,

      StartChildWorkflowExecutionFailedEventAttributes a,

      long notUsed) {

    StartChildWorkflowExecutionFailedEventAttributes.Builder updatedAttr =

        a.toBuilder()

            .setInitiatedEventId(data.initiatedEventId)

            .setWorkflowType(data.initiatedEvent.getWorkflowType())

            .setWorkflowId(data.initiatedEvent.getWorkflowId());

    if (!data.initiatedEvent.getNamespace().isEmpty()) {

      updatedAttr.setNamespace(data.initiatedEvent.getNamespace());

    }

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_FAILED)

            .setStartChildWorkflowExecutionFailedEventAttributes(updatedAttr.build())

            .build();

    ctx.addEvent(event);

  }

  private static void childWorkflowStarted(

      RequestContext ctx,

      ChildWorkflowData data,

      ChildWorkflowExecutionStartedEventAttributes a,

      long notUsed) {

    ChildWorkflowExecutionStartedEventAttributes updatedAttr =

        a.toBuilder().setInitiatedEventId(data.initiatedEventId).build();

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED)

            .setChildWorkflowExecutionStartedEventAttributes(updatedAttr)

            .build();

    long startedEventId = ctx.addEvent(event);

    ctx.onCommit(

        (historySize) -> {

          data.startedEventId = startedEventId;

          data.execution = updatedAttr.getWorkflowExecution();

        });

  }

  private static void childWorkflowCompleted(

      RequestContext ctx,

      ChildWorkflowData data,

      ChildWorkflowExecutionCompletedEventAttributes a,

      long notUsed) {

    ChildWorkflowExecutionCompletedEventAttributes updatedAttr =

        a.toBuilder()

            .setInitiatedEventId(data.initiatedEventId)

            .setStartedEventId(data.startedEventId)

            .build();

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED)

            .setChildWorkflowExecutionCompletedEventAttributes(updatedAttr)

            .build();

    ctx.addEvent(event);

  }

  private static void childWorkflowFailed(

      RequestContext ctx,

      ChildWorkflowData data,

      ChildWorkflowExecutionFailedEventAttributes a,

      long notUsed) {

    ChildWorkflowExecutionFailedEventAttributes.Builder updatedAttr =

        a.toBuilder()

            .setInitiatedEventId(data.initiatedEventId)

            .setStartedEventId(data.startedEventId)

            .setWorkflowExecution(data.execution)

            .setWorkflowType(data.initiatedEvent.getWorkflowType());

    if (!data.initiatedEvent.getNamespace().isEmpty()) {

      updatedAttr.setNamespace(data.initiatedEvent.getNamespace());

    }

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_FAILED)

            .setChildWorkflowExecutionFailedEventAttributes(updatedAttr.build())

            .build();

    ctx.addEvent(event);

  }

  private static void childWorkflowCanceled(

      RequestContext ctx,

      ChildWorkflowData data,

      ChildWorkflowExecutionCanceledEventAttributes a,

      long notUsed) {

    ChildWorkflowExecutionCanceledEventAttributes updatedAttr =

        a.toBuilder()

            .setInitiatedEventId(data.initiatedEventId)

            .setStartedEventId(data.startedEventId)

            .build();

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_CANCELED)

            .setChildWorkflowExecutionCanceledEventAttributes(updatedAttr)

            .build();

    ctx.addEvent(event);

  }

  private static void initiateChildWorkflow(

      RequestContext ctx,

      ChildWorkflowData data,

      StartChildWorkflowExecutionCommandAttributes d,

      long workflowTaskCompletedEventId) {

    StartChildWorkflowExecutionInitiatedEventAttributes.Builder a =

        StartChildWorkflowExecutionInitiatedEventAttributes.newBuilder()

            .setControl(d.getControl())

            .setInput(d.getInput())

            .setWorkflowTaskCompletedEventId(workflowTaskCompletedEventId)

            .setNamespace(d.getNamespace().isEmpty() ? ctx.getNamespace() : d.getNamespace())

            .setWorkflowExecutionTimeout(d.getWorkflowExecutionTimeout())

            .setWorkflowRunTimeout(d.getWorkflowRunTimeout())

            .setWorkflowTaskTimeout(d.getWorkflowTaskTimeout())

            .setTaskQueue(d.getTaskQueue())

            .setWorkflowId(d.getWorkflowId())

            .setWorkflowIdReusePolicy(d.getWorkflowIdReusePolicy())

            .setWorkflowType(d.getWorkflowType())

            .setCronSchedule(d.getCronSchedule())

            .setParentClosePolicy(d.getParentClosePolicy());

    if (d.hasHeader()) {

      a.setHeader(d.getHeader());

    }

    if (d.hasMemo()) {

      a.setMemo(d.getMemo());

    }

    if (d.hasRetryPolicy()) {

      a.setRetryPolicy(d.getRetryPolicy());

    }

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED)

            .setStartChildWorkflowExecutionInitiatedEventAttributes(a)

            .build();

    long initiatedEventId = ctx.addEvent(event);

    ctx.onCommit(

        (historySize) -> {

          data.initiatedEventId = initiatedEventId;

          data.initiatedEvent = a.build();

          StartWorkflowExecutionRequest.Builder startChild =

              StartWorkflowExecutionRequest.newBuilder()

                  .setRequestId(UUID.randomUUID().toString())

                  .setNamespace(d.getNamespace().isEmpty() ? ctx.getNamespace() : d.getNamespace())

                  .setWorkflowExecutionTimeout(d.getWorkflowExecutionTimeout())

                  .setWorkflowRunTimeout(d.getWorkflowRunTimeout())

                  .setWorkflowTaskTimeout(d.getWorkflowTaskTimeout())

                  .setTaskQueue(d.getTaskQueue())

                  .setWorkflowId(d.getWorkflowId())

                  .setWorkflowIdReusePolicy(d.getWorkflowIdReusePolicy())

                  .setWorkflowType(d.getWorkflowType())

                  .setCronSchedule(d.getCronSchedule());

          if (d.hasHeader()) {

            startChild.setHeader(d.getHeader());

          }

          if (d.hasSearchAttributes()) {

            startChild.setSearchAttributes(d.getSearchAttributes());

          }

          if (d.hasMemo()) {

            startChild.setMemo(d.getMemo());

          }

          if (d.hasRetryPolicy()) {

            startChild.setRetryPolicy(d.getRetryPolicy());

          }

          if (d.hasInput()) {

            startChild.setInput(d.getInput());

          }

          addStartChildTask(ctx, data, initiatedEventId, startChild.build());

        });

  }

  private static void addStartChildTask(

      RequestContext ctx,

      ChildWorkflowData data,

      long initiatedEventId,

      StartWorkflowExecutionRequest startChild) {

    ForkJoinPool.commonPool()

        .execute(

            () -> {

              try {

                data.service.startWorkflowExecutionImpl(

                    startChild,

                    java.time.Duration.ZERO,

                    Optional.of(ctx.getWorkflowMutableState()),

                    OptionalLong.of(data.initiatedEventId),

                    null);

              } catch (StatusRuntimeException e) {

                if (e.getStatus().getCode() == Status.Code.ALREADY_EXISTS) {

                  StartChildWorkflowExecutionFailedEventAttributes failRequest =

                      StartChildWorkflowExecutionFailedEventAttributes.newBuilder()

                          .setInitiatedEventId(initiatedEventId)

                          .setCause(

                              StartChildWorkflowExecutionFailedCause

                                  .START_CHILD_WORKFLOW_EXECUTION_FAILED_CAUSE_WORKFLOW_ALREADY_EXISTS)

                          .build();

                  try {

                    ctx.getWorkflowMutableState()

                        .failStartChildWorkflow(data.initiatedEvent.getWorkflowId(), failRequest);

                  } catch (Throwable ee) {

                    log.error("Unexpected failure inserting failStart for a child workflow", ee);

                  }

                } else {

                  log.error("Unexpected failure starting a child workflow", e);

                }

              } catch (Exception e) {

                log.error("Unexpected failure starting a child workflow", e);

              }

            });

  }

  private static void startWorkflow(

      RequestContext ctx, WorkflowData data, StartWorkflowExecutionRequest request, long notUsed) {

    if (Durations.compare(request.getWorkflowExecutionTimeout(), Durations.ZERO) < 0) {

      throw Status.INVALID_ARGUMENT

          .withDescription("negative workflowExecution timeout")

          .asRuntimeException();

    }

    if (Durations.compare(request.getWorkflowRunTimeout(), Durations.ZERO) < 0) {

      throw Status.INVALID_ARGUMENT

          .withDescription("negative workflowRun timeout")

          .asRuntimeException();

    }

    if (Durations.compare(request.getWorkflowTaskTimeout(), Durations.ZERO) < 0) {

      throw Status.INVALID_ARGUMENT

          .withDescription("negative workflowTaskTimeoutSeconds")

          .asRuntimeException();

    }

    WorkflowExecutionStartedEventAttributes.Builder a =

        WorkflowExecutionStartedEventAttributes.newBuilder()

            .setWorkflowType(request.getWorkflowType())

            .setWorkflowRunTimeout(request.getWorkflowRunTimeout())

            .setWorkflowTaskTimeout(request.getWorkflowTaskTimeout())

            .setWorkflowExecutionTimeout(request.getWorkflowExecutionTimeout())

            .setIdentity(request.getIdentity())

            .setInput(request.getInput())

            .setTaskQueue(request.getTaskQueue())

            .setAttempt(1);

    data.retryState.ifPresent(

        testServiceRetryState -> a.setAttempt(testServiceRetryState.getAttempt()));

    a.setFirstExecutionRunId(data.firstExecutionRunId);

    a.setOriginalExecutionRunId(data.originalExecutionRunId);

    data.continuedExecutionRunId.ifPresent(a::setContinuedExecutionRunId);

    if (data.lastCompletionResult != null) {

      a.setLastCompletionResult(data.lastCompletionResult);

    }

    data.lastFailure.ifPresent(a::setContinuedFailure);

    if (request.hasMemo()) {

      a.setMemo(request.getMemo());

    }

    if (request.hasSearchAttributes()) {

      a.setSearchAttributes((request.getSearchAttributes()));

    }

    if (request.hasHeader()) {

      a.setHeader(request.getHeader());

    }

    String cronSchedule = request.getCronSchedule();

    if (!cronSchedule.trim().isEmpty()) {

      try {

        CronUtils.parseCron(cronSchedule);

        a.setCronSchedule(cronSchedule);

      } catch (Exception e) {

        throw Status.INVALID_ARGUMENT

            .withDescription("Invalid cron expression \"" + cronSchedule + "\": " + e.getMessage())

            .withCause(e)

            .asRuntimeException();

      }

    }

    Optional<TestWorkflowMutableState> parent = ctx.getWorkflowMutableState().getParent();

    if (parent.isPresent()) {

      ExecutionId parentExecutionId = parent.get().getExecutionId();

      a.setParentWorkflowNamespace(parentExecutionId.getNamespace());

      a.setParentWorkflowExecution(parentExecutionId.getExecution());

    }

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED)

            .setWorkflowExecutionStartedEventAttributes(a)

            .build();

    ctx.addEvent(event);

  }

  private static void completeWorkflow(

      RequestContext ctx,

      WorkflowData data,

      CompleteWorkflowExecutionCommandAttributes d,

      long workflowTaskCompletedEventId) {

    WorkflowExecutionCompletedEventAttributes.Builder a =

        WorkflowExecutionCompletedEventAttributes.newBuilder()

            .setResult(d.getResult())

            .setWorkflowTaskCompletedEventId(workflowTaskCompletedEventId);

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED)

            .setWorkflowExecutionCompletedEventAttributes(a)

            .build();

    ctx.addEvent(event);

  }

  private static void continueAsNewWorkflow(

      RequestContext ctx,

      WorkflowData data,

      ContinueAsNewWorkflowExecutionCommandAttributes d,

      long workflowTaskCompletedEventId) {

    StartWorkflowExecutionRequest sr = ctx.getWorkflowMutableState().getStartRequest();

    WorkflowExecutionContinuedAsNewEventAttributes.Builder a =

        WorkflowExecutionContinuedAsNewEventAttributes.newBuilder();

    a.setInput(d.getInput());

    if (d.hasHeader()) {

      a.setHeader(d.getHeader());

    }

    if (Durations.compare(d.getWorkflowRunTimeout(), Durations.ZERO) > 0) {

      a.setWorkflowRunTimeout(d.getWorkflowRunTimeout());

    } else {

      a.setWorkflowRunTimeout(sr.getWorkflowRunTimeout());

    }

    if (d.hasTaskQueue()) {

      a.setTaskQueue(d.getTaskQueue());

    } else {

      a.setTaskQueue(sr.getTaskQueue());

    }

    if (d.hasWorkflowType()) {

      a.setWorkflowType(d.getWorkflowType());

    } else {

      a.setWorkflowType(sr.getWorkflowType());

    }

    if (Durations.compare(d.getWorkflowTaskTimeout(), Durations.ZERO) > 0) {

      a.setWorkflowTaskTimeout(d.getWorkflowTaskTimeout());

    } else {

      a.setWorkflowTaskTimeout(sr.getWorkflowTaskTimeout());

    }

    a.setWorkflowTaskCompletedEventId(workflowTaskCompletedEventId);

    a.setBackoffStartInterval(d.getBackoffStartInterval());

    if (d.hasLastCompletionResult()) {

      a.setLastCompletionResult(d.getLastCompletionResult());

    }

    if (d.hasFailure()) {

      a.setFailure(d.getFailure());

    }

    a.setNewExecutionRunId(UUID.randomUUID().toString());

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW)

            .setWorkflowExecutionContinuedAsNewEventAttributes(a)

            .build();

    ctx.addEvent(event);

  }

  private static void failWorkflow(

      RequestContext ctx,

      WorkflowData data,

      FailWorkflowExecutionCommandAttributes d,

      long workflowTaskCompletedEventId) {

    WorkflowExecutionFailedEventAttributes.Builder a =

        WorkflowExecutionFailedEventAttributes.newBuilder()

            .setWorkflowTaskCompletedEventId(workflowTaskCompletedEventId);

    if (d.hasFailure()) {

      a.setFailure(d.getFailure());

    }

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED)

            .setWorkflowExecutionFailedEventAttributes(a)

            .build();

    ctx.addEvent(event);

  }

  private static void timeoutWorkflow(

      RequestContext ctx, WorkflowData data, RetryState retryState, long notUsed) {

    WorkflowExecutionTimedOutEventAttributes.Builder a =

        WorkflowExecutionTimedOutEventAttributes.newBuilder().setRetryState(retryState);

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT)

            .setWorkflowExecutionTimedOutEventAttributes(a)

            .build();

    ctx.addEvent(event);

  }

  private static void cancelWorkflow(

      RequestContext ctx,

      WorkflowData data,

      CancelWorkflowExecutionCommandAttributes d,

      long workflowTaskCompletedEventId) {

    WorkflowExecutionCanceledEventAttributes.Builder a =

        WorkflowExecutionCanceledEventAttributes.newBuilder()

            .setDetails(d.getDetails())

            .setWorkflowTaskCompletedEventId(workflowTaskCompletedEventId);

    HistoryEvent event =

        HistoryEvent.newBuilder()

            .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CANCELED)

            .setWorkflowExecutionCanceledEventAttributes(a)

            .build();

    ctx.addEvent(event);

  }

  private static void terminateWorkflow(

      RequestContext ctx,

      WorkflowData data,