#159

pending completion

Build # #159

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Fix premature triggering of eventLoop in case of activity cancellation (#1691)

Issue #1558

Run Details

26 of 26 new or added lines in 2 files covered. (100.0%)

17016 of 20841 relevant lines covered (81.65%)

0.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.4

/temporal-sdk/src/main/java/io/temporal/internal/statemachines/ActivityStateMachine.java

/*
 * 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.statemachines;

import io.temporal.activity.ActivityCancellationType;
import io.temporal.api.command.v1.Command;
import io.temporal.api.command.v1.RequestCancelActivityTaskCommandAttributes;
import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributes;
import io.temporal.api.common.v1.ActivityType;
import io.temporal.api.common.v1.Payloads;
import io.temporal.api.enums.v1.CommandType;
import io.temporal.api.enums.v1.EventType;
import io.temporal.api.failure.v1.ActivityFailureInfo;
import io.temporal.api.failure.v1.CanceledFailureInfo;
import io.temporal.api.failure.v1.Failure;
import io.temporal.api.history.v1.ActivityTaskCanceledEventAttributes;
import io.temporal.api.history.v1.ActivityTaskCompletedEventAttributes;
import io.temporal.api.history.v1.ActivityTaskFailedEventAttributes;
import io.temporal.api.history.v1.ActivityTaskTimedOutEventAttributes;
import io.temporal.workflow.Functions;
import java.util.Optional;
import javax.annotation.Nonnull;

final class ActivityStateMachine
    extends EntityStateMachineInitialCommand<
        ActivityStateMachine.State, ActivityStateMachine.ExplicitEvent, ActivityStateMachine> {

  static final String ACTIVITY_FAILED_MESSAGE = "Activity task failed";

  static final String ACTIVITY_TIMED_OUT_MESSAGE = "Activity task timed out";

  static final String ACTIVITY_CANCELED_MESSAGE = "Activity canceled";

  private static final String JAVA_SDK = "JavaSDK";

  private final String activityId;
  private final ActivityType activityType;
  private final ActivityCancellationType cancellationType;

  private final Functions.Proc2<Optional<Payloads>, FailureResult> completionCallback;

  private ExecuteActivityParameters parameters;

  private long startedCommandEventId;

  enum ExplicitEvent {
    SCHEDULE,
    CANCEL
  }

  enum State {
    CREATED,
    SCHEDULE_COMMAND_CREATED,
    SCHEDULED_EVENT_RECORDED,
    STARTED,
    COMPLETED,
    FAILED,
    TIMED_OUT,
    CANCELED,
    SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
    SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED,
    STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
    STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
  }

  public static final StateMachineDefinition<State, ExplicitEvent, ActivityStateMachine>
      STATE_MACHINE_DEFINITION =
          StateMachineDefinition.<State, ExplicitEvent, ActivityStateMachine>newInstance(
                  "Activity",
                  State.CREATED,
                  State.COMPLETED,
                  State.FAILED,
                  State.TIMED_OUT,
                  State.CANCELED)
              .add(
                  State.CREATED,
                  ExplicitEvent.SCHEDULE,
                  State.SCHEDULE_COMMAND_CREATED,
                  ActivityStateMachine::createScheduleActivityTaskCommand)
              .add(
                  State.SCHEDULE_COMMAND_CREATED,
                  CommandType.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK,
                  State.SCHEDULE_COMMAND_CREATED)
              .add(
                  State.SCHEDULE_COMMAND_CREATED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED,
                  State.SCHEDULED_EVENT_RECORDED,
                  ActivityStateMachine::setInitialCommandEventId)
              .add(
                  State.SCHEDULE_COMMAND_CREATED,
                  ExplicitEvent.CANCEL,
                  State.CANCELED,
                  ActivityStateMachine::cancelCommandNotifyCanceledImmediately)
              .add(
                  State.SCHEDULED_EVENT_RECORDED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_STARTED,
                  State.STARTED,
                  ActivityStateMachine::setStartedCommandEventId)
              .add(
                  State.SCHEDULED_EVENT_RECORDED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
                  State.TIMED_OUT,
                  ActivityStateMachine::notifyTimedOut)
              .add(
                  State.SCHEDULED_EVENT_RECORDED,
                  ExplicitEvent.CANCEL,
                  State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  ActivityStateMachine::createRequestCancelActivityTaskCommand)
              .add(
                  State.STARTED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_COMPLETED,
                  State.COMPLETED,
                  ActivityStateMachine::notifyCompleted)
              .add(
                  State.STARTED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_FAILED,
                  State.FAILED,
                  ActivityStateMachine::notifyFailed)
              .add(
                  State.STARTED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
                  State.TIMED_OUT,
                  ActivityStateMachine::notifyTimedOut)
              .add(
                  State.STARTED,
                  ExplicitEvent.CANCEL,
                  State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  ActivityStateMachine::createRequestCancelActivityTaskCommand)
              .add(
                  State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  CommandType.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK,
                  State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  ActivityStateMachine::notifyCanceledIfTryCancelImmediately)
              .add(
                  State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED,
                  State.SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED)
              /*
              These state transitions are not possible.
              It looks like it is valid when an event, handling of which requests activity
              cancellation, precedes EVENT_TYPE_ACTIVITY_TASK_STARTED event.
              But as all code execution happens in the event loop the STARTED event is
              applied to the state machine (as it is done for all command events before
              the event loop invocation) before the cancellation request.
              .add(
                  State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_STARTED,
                  State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED)
               This one is not possible for similar reason. The timeout is delivered
               before the event loop execution.
              .add(
                  State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
                  State.TIMED_OUT,
                  ActivityStateMachine::cancelCommandNotifyTimedOut)
                   */
              .add(
                  State.SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_CANCELED,
                  State.CANCELED,
                  ActivityStateMachine::notifyCanceledFromEvent)
              .add(
                  State.SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_STARTED,
                  State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED)
              .add(
                  State.SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
                  State.TIMED_OUT,
                  ActivityStateMachine::notifyTimedOut)
              .add(
                  State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  CommandType.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK,
                  State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED)
              .add(
                  State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED,
                  State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
                  ActivityStateMachine::notifyCanceledIfTryCancelFromEvent)
              /*
              These state transitions are not possible.
              It looks like it is valid when an event, handling of which requests activity
              cancellation, precedes EVENT_TYPE_ACTIVITY_TASK_[COMPLETED|FAILED|TIMED_OUT] event.
              But as all code execution happens in the event loop the completion event is
              applied to the state machine (as it is done for all command events before
              the event loop invocation) before the cancellation request.
              .add(
                  State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_COMPLETED,
                  State.COMPLETED,
                  ActivityStateMachine::cancelCommandNotifyCompleted)
              .add(
                  State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_FAILED,
                  State.FAILED,
                  ActivityStateMachine::cancelCommandNotifyFailed)
              .add(
                  State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
                  State.TIMED_OUT,
                  ActivityStateMachine::cancelCommandNotifyTimedOut)
                   */
              .add(
                  State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_FAILED,
                  State.FAILED,
                  ActivityStateMachine::notifyFailed)
              .add(
                  State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_COMPLETED,
                  State.COMPLETED,
                  ActivityStateMachine::notifyCompleted)
              .add(
                  State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
                  State.TIMED_OUT,
                  ActivityStateMachine::notifyTimedOut)
              .add(
                  State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
                  EventType.EVENT_TYPE_ACTIVITY_TASK_CANCELED,
                  State.CANCELED,
                  ActivityStateMachine::notifyCancellationFromEvent);

  /**
   * @param parameters attributes used to schedule an activity
   * @param completionCallback one of ActivityTaskCompletedEvent, ActivityTaskFailedEvent,
   *     ActivityTaskTimedOutEvent, ActivityTaskCanceledEvents
   * @param commandSink sink to send commands
   * @return an instance of ActivityCommands
   */
  public static ActivityStateMachine newInstance(
      ExecuteActivityParameters parameters,
      Functions.Proc2<Optional<Payloads>, FailureResult> completionCallback,
      Functions.Proc1<CancellableCommand> commandSink,
      Functions.Proc1<StateMachine> stateMachineSink) {
    return new ActivityStateMachine(parameters, completionCallback, commandSink, stateMachineSink);
  }

  private ActivityStateMachine(
      ExecuteActivityParameters parameters,
      Functions.Proc2<Optional<Payloads>, FailureResult> completionCallback,
      Functions.Proc1<CancellableCommand> commandSink,
      Functions.Proc1<StateMachine> stateMachineSink) {
    super(STATE_MACHINE_DEFINITION, commandSink, stateMachineSink);
    this.parameters = parameters;
    ScheduleActivityTaskCommandAttributes.Builder scheduleAttr = parameters.getAttributes();
    this.activityId = scheduleAttr.getActivityId();
    this.activityType = scheduleAttr.getActivityType();
    this.cancellationType = parameters.getCancellationType();
    this.completionCallback = completionCallback;
    explicitEvent(ExplicitEvent.SCHEDULE);
  }

  public void createScheduleActivityTaskCommand() {
    addCommand(
        Command.newBuilder()
            .setCommandType(CommandType.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK)
            .setScheduleActivityTaskCommandAttributes(parameters.getAttributes())
            .build());
  }

  private void setStartedCommandEventId() {
    startedCommandEventId = currentEvent.getEventId();
  }

  public void cancel() {
    if (cancellationType == ActivityCancellationType.ABANDON) {
      notifyCanceled(false);
    } else if (!isFinalState()) {
      explicitEvent(ExplicitEvent.CANCEL);
    }
  }

  // *Immediately versions don't wait for a matching command, underlying callback will trigger an
  // event loop so the workflow can make progress, because promise gets filled.

  /**
   * {@link CommandType#COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK} command is not yet left to the server.
   * Cancel it in place and immediately notify the workflow code.
   */
  private void cancelCommandNotifyCanceledImmediately() {
    cancelCommand();
    // TODO With {@link ActivityCancellationType#ABANDON} we shouldn't even get here as it gets
    //  handled in #cancel.
    //  It's a code path for TRY_CANCEL and WAIT_CANCELLATION_COMPLETED only.
    //  Was the original design to cancel a not-yet-sent
    //  {@link CommandType#COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK} in case of ABANDON too?
    if (cancellationType != ActivityCancellationType.ABANDON) {
      notifyCanceled(false);
    }
  }

  /**
   * Workflow code doesn't need to wait for the cancellation event if {@link
   * ActivityCancellationType#TRY_CANCEL}, immediately notify the workflow code.
   */
  private void notifyCanceledIfTryCancelImmediately() {
    if (cancellationType == ActivityCancellationType.TRY_CANCEL) {
      notifyCanceled(false);
    }
  }

  // *FromEvent versions will not trigger event loop as they need to wait for all events to be
  // applied before and there will be WorkflowTaskStarted to trigger the event loop.

  /**
   * if {@link EventType#EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED} is observed, notify the workflow
   * code if {@link ActivityCancellationType#TRY_CANCEL}, this mode doesn't need a confirmation of
   * cancellation.
   */
  private void notifyCanceledIfTryCancelFromEvent() {
    if (cancellationType == ActivityCancellationType.TRY_CANCEL) {
      notifyCanceled(true);
    }
  }

  /**
   * Notify workflow code of the cancellation from the {@link
   * EventType#EVENT_TYPE_ACTIVITY_TASK_CANCELED} event.
   *
   * <p>There is no harm in notifying {@link ActivityCancellationType#TRY_CANCEL} again, but it
   * should not be needed as it should be already done by {@link
   * #notifyCanceledIfTryCancelFromEvent} as there should be no {@link
   * EventType#EVENT_TYPE_ACTIVITY_TASK_CANCELED} without {@link
   * EventType#EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED}.
   */
  private void notifyCanceledFromEvent() {
    notifyCanceled(true);
  }

  private void notifyCanceled(boolean fromEvent) {
    Failure canceledFailure =
        Failure.newBuilder()
            .setSource(JAVA_SDK)
            .setCanceledFailureInfo(CanceledFailureInfo.getDefaultInstance())
            .build();
    ActivityFailureInfo activityFailureInfo =
        ActivityFailureInfo.newBuilder()
            .setActivityId(activityId)
            .setActivityType(activityType)
            .setIdentity("workflow")
            .setScheduledEventId(getInitialCommandEventId())
            .setStartedEventId(startedCommandEventId)
            .build();
    Failure failure =
        Failure.newBuilder()
            .setActivityFailureInfo(activityFailureInfo)
            .setCause(canceledFailure)
            .setMessage(ACTIVITY_CANCELED_MESSAGE)
            .build();
    completionCallback.apply(Optional.empty(), new FailureResult(failure, fromEvent));
  }

  private void notifyCompleted() {
    ActivityTaskCompletedEventAttributes completedAttr =
        currentEvent.getActivityTaskCompletedEventAttributes();
    Optional<Payloads> result =
        completedAttr.hasResult() ? Optional.of(completedAttr.getResult()) : Optional.empty();
    completionCallback.apply(result, null);
  }

  private void notifyFailed() {
    ActivityTaskFailedEventAttributes failed = currentEvent.getActivityTaskFailedEventAttributes();
    ActivityFailureInfo failureInfo =
        ActivityFailureInfo.newBuilder()
            .setActivityId(activityId)
            .setActivityType(activityType)
            .setIdentity(failed.getIdentity())
            .setRetryState(failed.getRetryState())
            .setScheduledEventId(failed.getScheduledEventId())
            .setStartedEventId(failed.getStartedEventId())
            .build();
    Failure failure =
        Failure.newBuilder()
            .setActivityFailureInfo(failureInfo)
            .setCause(failed.getFailure())
            .setMessage(ACTIVITY_FAILED_MESSAGE)
            .build();
    completionCallback.apply(Optional.empty(), new FailureResult(failure, true));
  }

  private void notifyTimedOut() {
    ActivityTaskTimedOutEventAttributes timedOut =
        currentEvent.getActivityTaskTimedOutEventAttributes();

    ActivityFailureInfo failureInfo =
        ActivityFailureInfo.newBuilder()
            .setActivityId(activityId)
            .setActivityType(activityType)
            .setRetryState(timedOut.getRetryState())
            .setScheduledEventId(timedOut.getScheduledEventId())
            .setStartedEventId(timedOut.getStartedEventId())
            .build();
    Failure failure =
        Failure.newBuilder()
            .setActivityFailureInfo(failureInfo)
            .setCause(timedOut.getFailure())
            .setMessage(ACTIVITY_TIMED_OUT_MESSAGE)
            .build();
    completionCallback.apply(Optional.empty(), new FailureResult(failure, true));
  }

  private void notifyCancellationFromEvent() {
    if (cancellationType == ActivityCancellationType.WAIT_CANCELLATION_COMPLETED) {
      ActivityTaskCanceledEventAttributes canceledAttr =
          currentEvent.getActivityTaskCanceledEventAttributes();
      Failure canceledFailure =
          Failure.newBuilder()
              .setSource(JAVA_SDK)
              .setCanceledFailureInfo(
                  CanceledFailureInfo.newBuilder().setDetails(canceledAttr.getDetails()))
              .build();

      ActivityFailureInfo failureInfo =
          ActivityFailureInfo.newBuilder()
              .setActivityId(activityId)
              .setActivityType(activityType)
              .setScheduledEventId(canceledAttr.getScheduledEventId())
              .setStartedEventId(canceledAttr.getStartedEventId())
              .build();
      Failure failure =
          Failure.newBuilder()
              .setActivityFailureInfo(failureInfo)
              .setCause(canceledFailure)
              .setMessage(ACTIVITY_CANCELED_MESSAGE)
              .build();

      completionCallback.apply(Optional.empty(), new FailureResult(failure, true));
    }
  }

  private void createRequestCancelActivityTaskCommand() {
    addCommand(
        Command.newBuilder()
            .setCommandType(CommandType.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK)
            .setRequestCancelActivityTaskCommandAttributes(
                RequestCancelActivityTaskCommandAttributes.newBuilder()
                    .setScheduledEventId(getInitialCommandEventId()))
            .build());
    parameters = null; // avoid retaining large input for the duration of the activity
  }

  public static class FailureResult {
    private final @Nonnull Failure failure;
    private final boolean fromEvent;

    public FailureResult(@Nonnull Failure failure, boolean fromEvent) {
      this.failure = failure;
      this.fromEvent = fromEvent;
    }

    @Nonnull
    public Failure getFailure() {
      return failure;
    }

    /**
     * @return true if this failure is created from the event during event <-> command matching
     *     phase.
     */
    public boolean isFromEvent() {
      return fromEvent;
    }
  }
}

1	/*
2	* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
3	*
4	* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
5	*
6	* Modifications copyright (C) 2017 Uber Technologies, Inc.
7	*
8	* Licensed under the Apache License, Version 2.0 (the "License");
9	* you may not use this material except in compliance with the License.
10	* You may obtain a copy of the License at
11	*
12	* http://www.apache.org/licenses/LICENSE-2.0
13	*
14	* Unless required by applicable law or agreed to in writing, software
15	* distributed under the License is distributed on an "AS IS" BASIS,
16	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17	* See the License for the specific language governing permissions and
18	* limitations under the License.
19	*/
20
21	package io.temporal.internal.statemachines;
22
23	import io.temporal.activity.ActivityCancellationType;
24	import io.temporal.api.command.v1.Command;
25	import io.temporal.api.command.v1.RequestCancelActivityTaskCommandAttributes;
26	import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributes;
27	import io.temporal.api.common.v1.ActivityType;
28	import io.temporal.api.common.v1.Payloads;
29	import io.temporal.api.enums.v1.CommandType;
30	import io.temporal.api.enums.v1.EventType;
31	import io.temporal.api.failure.v1.ActivityFailureInfo;
32	import io.temporal.api.failure.v1.CanceledFailureInfo;
33	import io.temporal.api.failure.v1.Failure;
34	import io.temporal.api.history.v1.ActivityTaskCanceledEventAttributes;
35	import io.temporal.api.history.v1.ActivityTaskCompletedEventAttributes;
36	import io.temporal.api.history.v1.ActivityTaskFailedEventAttributes;
37	import io.temporal.api.history.v1.ActivityTaskTimedOutEventAttributes;
38	import io.temporal.workflow.Functions;
39	import java.util.Optional;
40	import javax.annotation.Nonnull;
41
42	final class ActivityStateMachine
43	extends EntityStateMachineInitialCommand<
44	ActivityStateMachine.State, ActivityStateMachine.ExplicitEvent, ActivityStateMachine> {
45
46	static final String ACTIVITY_FAILED_MESSAGE = "Activity task failed";
47
48	static final String ACTIVITY_TIMED_OUT_MESSAGE = "Activity task timed out";
49
50	static final String ACTIVITY_CANCELED_MESSAGE = "Activity canceled";
51
52	private static final String JAVA_SDK = "JavaSDK";
53
54	private final String activityId;
55	private final ActivityType activityType;
56	private final ActivityCancellationType cancellationType;
57
58	private final Functions.Proc2<Optional<Payloads>, FailureResult> completionCallback;
59
60	private ExecuteActivityParameters parameters;
61
62	private long startedCommandEventId;
63
64	enum ExplicitEvent {	1✔
65	SCHEDULE,	1✔
66	CANCEL	1✔
67	}
68
69	enum State {	1✔
70	CREATED,	1✔
71	SCHEDULE_COMMAND_CREATED,	1✔
72	SCHEDULED_EVENT_RECORDED,	1✔
73	STARTED,	1✔
74	COMPLETED,	1✔
75	FAILED,	1✔
76	TIMED_OUT,	1✔
77	CANCELED,	1✔
78	SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,	1✔
79	SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED,	1✔
80	STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,	1✔
81	STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,	1✔
82	}
83
84	public static final StateMachineDefinition<State, ExplicitEvent, ActivityStateMachine>
85	STATE_MACHINE_DEFINITION =	1✔
86	StateMachineDefinition.<State, ExplicitEvent, ActivityStateMachine>newInstance(	1✔
87	"Activity",
88	State.CREATED,
89	State.COMPLETED,
90	State.FAILED,
91	State.TIMED_OUT,
92	State.CANCELED)
93	.add(	1✔
94	State.CREATED,
95	ExplicitEvent.SCHEDULE,
96	State.SCHEDULE_COMMAND_CREATED,
97	ActivityStateMachine::createScheduleActivityTaskCommand)
98	.add(	1✔
99	State.SCHEDULE_COMMAND_CREATED,
100	CommandType.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK,
101	State.SCHEDULE_COMMAND_CREATED)
102	.add(	1✔
103	State.SCHEDULE_COMMAND_CREATED,
104	EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED,
105	State.SCHEDULED_EVENT_RECORDED,
106	ActivityStateMachine::setInitialCommandEventId)
107	.add(	1✔
108	State.SCHEDULE_COMMAND_CREATED,
109	ExplicitEvent.CANCEL,
110	State.CANCELED,
111	ActivityStateMachine::cancelCommandNotifyCanceledImmediately)
112	.add(	1✔
113	State.SCHEDULED_EVENT_RECORDED,
114	EventType.EVENT_TYPE_ACTIVITY_TASK_STARTED,
115	State.STARTED,
116	ActivityStateMachine::setStartedCommandEventId)
117	.add(	1✔
118	State.SCHEDULED_EVENT_RECORDED,
119	EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
120	State.TIMED_OUT,
121	ActivityStateMachine::notifyTimedOut)
122	.add(	1✔
123	State.SCHEDULED_EVENT_RECORDED,
124	ExplicitEvent.CANCEL,
125	State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
126	ActivityStateMachine::createRequestCancelActivityTaskCommand)
127	.add(	1✔
128	State.STARTED,
129	EventType.EVENT_TYPE_ACTIVITY_TASK_COMPLETED,
130	State.COMPLETED,
131	ActivityStateMachine::notifyCompleted)
132	.add(	1✔
133	State.STARTED,
134	EventType.EVENT_TYPE_ACTIVITY_TASK_FAILED,
135	State.FAILED,
136	ActivityStateMachine::notifyFailed)
137	.add(	1✔
138	State.STARTED,
139	EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
140	State.TIMED_OUT,
141	ActivityStateMachine::notifyTimedOut)
142	.add(	1✔
143	State.STARTED,
144	ExplicitEvent.CANCEL,
145	State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
146	ActivityStateMachine::createRequestCancelActivityTaskCommand)
147	.add(	1✔
148	State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
149	CommandType.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK,
150	State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
151	ActivityStateMachine::notifyCanceledIfTryCancelImmediately)
152	.add(	1✔
153	State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
154	EventType.EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED,
155	State.SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED)
156	/*
157	These state transitions are not possible.
158	It looks like it is valid when an event, handling of which requests activity
159	cancellation, precedes EVENT_TYPE_ACTIVITY_TASK_STARTED event.
160	But as all code execution happens in the event loop the STARTED event is
161	applied to the state machine (as it is done for all command events before
162	the event loop invocation) before the cancellation request.
163	.add(
164	State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
165	EventType.EVENT_TYPE_ACTIVITY_TASK_STARTED,
166	State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED)
167	This one is not possible for similar reason. The timeout is delivered
168	before the event loop execution.
169	.add(
170	State.SCHEDULED_ACTIVITY_CANCEL_COMMAND_CREATED,
171	EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
172	State.TIMED_OUT,
173	ActivityStateMachine::cancelCommandNotifyTimedOut)
174	*/
175	.add(	1✔
176	State.SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED,
177	EventType.EVENT_TYPE_ACTIVITY_TASK_CANCELED,
178	State.CANCELED,
179	ActivityStateMachine::notifyCanceledFromEvent)
180	.add(	1✔
181	State.SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED,
182	EventType.EVENT_TYPE_ACTIVITY_TASK_STARTED,
183	State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED)
184	.add(	1✔
185	State.SCHEDULED_ACTIVITY_CANCEL_EVENT_RECORDED,
186	EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
187	State.TIMED_OUT,
188	ActivityStateMachine::notifyTimedOut)
189	.add(	1✔
190	State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
191	CommandType.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK,
192	State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED)
193	.add(	1✔
194	State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
195	EventType.EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED,
196	State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
197	ActivityStateMachine::notifyCanceledIfTryCancelFromEvent)
198	/*
199	These state transitions are not possible.
200	It looks like it is valid when an event, handling of which requests activity
201	cancellation, precedes EVENT_TYPE_ACTIVITY_TASK_[COMPLETED\|FAILED\|TIMED_OUT] event.
202	But as all code execution happens in the event loop the completion event is
203	applied to the state machine (as it is done for all command events before
204	the event loop invocation) before the cancellation request.
205	.add(
206	State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
207	EventType.EVENT_TYPE_ACTIVITY_TASK_COMPLETED,
208	State.COMPLETED,
209	ActivityStateMachine::cancelCommandNotifyCompleted)
210	.add(
211	State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
212	EventType.EVENT_TYPE_ACTIVITY_TASK_FAILED,
213	State.FAILED,
214	ActivityStateMachine::cancelCommandNotifyFailed)
215	.add(
216	State.STARTED_ACTIVITY_CANCEL_COMMAND_CREATED,
217	EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
218	State.TIMED_OUT,
219	ActivityStateMachine::cancelCommandNotifyTimedOut)
220	*/
221	.add(	1✔
222	State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
223	EventType.EVENT_TYPE_ACTIVITY_TASK_FAILED,
224	State.FAILED,
225	ActivityStateMachine::notifyFailed)
226	.add(	1✔
227	State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
228	EventType.EVENT_TYPE_ACTIVITY_TASK_COMPLETED,
229	State.COMPLETED,
230	ActivityStateMachine::notifyCompleted)
231	.add(	1✔
232	State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
233	EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT,
234	State.TIMED_OUT,
235	ActivityStateMachine::notifyTimedOut)
236	.add(	1✔
237	State.STARTED_ACTIVITY_CANCEL_EVENT_RECORDED,
238	EventType.EVENT_TYPE_ACTIVITY_TASK_CANCELED,
239	State.CANCELED,
240	ActivityStateMachine::notifyCancellationFromEvent);
241
242	/**
243	* @param parameters attributes used to schedule an activity
244	* @param completionCallback one of ActivityTaskCompletedEvent, ActivityTaskFailedEvent,
245	* ActivityTaskTimedOutEvent, ActivityTaskCanceledEvents
246	* @param commandSink sink to send commands
247	* @return an instance of ActivityCommands
248	*/
249	public static ActivityStateMachine newInstance(
250	ExecuteActivityParameters parameters,
251	Functions.Proc2<Optional<Payloads>, FailureResult> completionCallback,
252	Functions.Proc1<CancellableCommand> commandSink,
253	Functions.Proc1<StateMachine> stateMachineSink) {
254	return new ActivityStateMachine(parameters, completionCallback, commandSink, stateMachineSink);	1✔
255	}
256
257	private ActivityStateMachine(
258	ExecuteActivityParameters parameters,
259	Functions.Proc2<Optional<Payloads>, FailureResult> completionCallback,
260	Functions.Proc1<CancellableCommand> commandSink,
261	Functions.Proc1<StateMachine> stateMachineSink) {
262	super(STATE_MACHINE_DEFINITION, commandSink, stateMachineSink);	1✔
263	this.parameters = parameters;	1✔
264	ScheduleActivityTaskCommandAttributes.Builder scheduleAttr = parameters.getAttributes();	1✔
265	this.activityId = scheduleAttr.getActivityId();	1✔
266	this.activityType = scheduleAttr.getActivityType();	1✔
267	this.cancellationType = parameters.getCancellationType();	1✔
268	this.completionCallback = completionCallback;	1✔
269	explicitEvent(ExplicitEvent.SCHEDULE);	1✔
270	}	1✔
271
272	public void createScheduleActivityTaskCommand() {
273	addCommand(	1✔
274	Command.newBuilder()	1✔
275	.setCommandType(CommandType.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK)	1✔
276	.setScheduleActivityTaskCommandAttributes(parameters.getAttributes())	1✔
277	.build());	1✔
278	}	1✔
279
280	private void setStartedCommandEventId() {
281	startedCommandEventId = currentEvent.getEventId();	1✔
282	}	1✔
283
284	public void cancel() {
285	if (cancellationType == ActivityCancellationType.ABANDON) {	1✔
286	notifyCanceled(false);	1✔
287	} else if (!isFinalState()) {	1✔
288	explicitEvent(ExplicitEvent.CANCEL);	1✔
289	}
290	}	1✔
291
292	// *Immediately versions don't wait for a matching command, underlying callback will trigger an
293	// event loop so the workflow can make progress, because promise gets filled.
294
295	/**
296	* {@link CommandType#COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK} command is not yet left to the server.
297	* Cancel it in place and immediately notify the workflow code.
298	*/
299	private void cancelCommandNotifyCanceledImmediately() {
300	cancelCommand();	1✔
301	// TODO With {@link ActivityCancellationType#ABANDON} we shouldn't even get here as it gets
302	// handled in #cancel.
303	// It's a code path for TRY_CANCEL and WAIT_CANCELLATION_COMPLETED only.
304	// Was the original design to cancel a not-yet-sent
305	// {@link CommandType#COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK} in case of ABANDON too?
306	if (cancellationType != ActivityCancellationType.ABANDON) {	1✔
307	notifyCanceled(false);	1✔
308	}
309	}	1✔
310
311	/**
312	* Workflow code doesn't need to wait for the cancellation event if {@link
313	* ActivityCancellationType#TRY_CANCEL}, immediately notify the workflow code.
314	*/
315	private void notifyCanceledIfTryCancelImmediately() {
316	if (cancellationType == ActivityCancellationType.TRY_CANCEL) {	1✔
317	notifyCanceled(false);	1✔
318	}
319	}	1✔
320
321	// *FromEvent versions will not trigger event loop as they need to wait for all events to be
322	// applied before and there will be WorkflowTaskStarted to trigger the event loop.
323
324	/**
325	* if {@link EventType#EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED} is observed, notify the workflow
326	* code if {@link ActivityCancellationType#TRY_CANCEL}, this mode doesn't need a confirmation of
327	* cancellation.
328	*/
329	private void notifyCanceledIfTryCancelFromEvent() {
330	if (cancellationType == ActivityCancellationType.TRY_CANCEL) {	1✔
331	notifyCanceled(true);	×
332	}
333	}	1✔
334
335	/**
336	* Notify workflow code of the cancellation from the {@link
337	* EventType#EVENT_TYPE_ACTIVITY_TASK_CANCELED} event.
338	*
339	* <p>There is no harm in notifying {@link ActivityCancellationType#TRY_CANCEL} again, but it
340	* should not be needed as it should be already done by {@link
341	* #notifyCanceledIfTryCancelFromEvent} as there should be no {@link
342	* EventType#EVENT_TYPE_ACTIVITY_TASK_CANCELED} without {@link
343	* EventType#EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED}.
344	*/
345	private void notifyCanceledFromEvent() {
346	notifyCanceled(true);	1✔
347	}	1✔
348
349	private void notifyCanceled(boolean fromEvent) {
350	Failure canceledFailure =
351	Failure.newBuilder()	1✔
352	.setSource(JAVA_SDK)	1✔
353	.setCanceledFailureInfo(CanceledFailureInfo.getDefaultInstance())	1✔
354	.build();	1✔
355	ActivityFailureInfo activityFailureInfo =
356	ActivityFailureInfo.newBuilder()	1✔
357	.setActivityId(activityId)	1✔
358	.setActivityType(activityType)	1✔
359	.setIdentity("workflow")	1✔
360	.setScheduledEventId(getInitialCommandEventId())	1✔
361	.setStartedEventId(startedCommandEventId)	1✔
362	.build();	1✔
363	Failure failure =
364	Failure.newBuilder()	1✔
365	.setActivityFailureInfo(activityFailureInfo)	1✔
366	.setCause(canceledFailure)	1✔
367	.setMessage(ACTIVITY_CANCELED_MESSAGE)	1✔
368	.build();	1✔
369	completionCallback.apply(Optional.empty(), new FailureResult(failure, fromEvent));	1✔
370	}	1✔
371
372	private void notifyCompleted() {
373	ActivityTaskCompletedEventAttributes completedAttr =	1✔
374	currentEvent.getActivityTaskCompletedEventAttributes();	1✔
375	Optional<Payloads> result =
376	completedAttr.hasResult() ? Optional.of(completedAttr.getResult()) : Optional.empty();	1✔
377	completionCallback.apply(result, null);	1✔
378	}	1✔
379
380	private void notifyFailed() {
381	ActivityTaskFailedEventAttributes failed = currentEvent.getActivityTaskFailedEventAttributes();	1✔
382	ActivityFailureInfo failureInfo =
383	ActivityFailureInfo.newBuilder()	1✔
384	.setActivityId(activityId)	1✔
385	.setActivityType(activityType)	1✔
386	.setIdentity(failed.getIdentity())	1✔
387	.setRetryState(failed.getRetryState())	1✔
388	.setScheduledEventId(failed.getScheduledEventId())	1✔
389	.setStartedEventId(failed.getStartedEventId())	1✔
390	.build();	1✔
391	Failure failure =
392	Failure.newBuilder()	1✔
393	.setActivityFailureInfo(failureInfo)	1✔
394	.setCause(failed.getFailure())	1✔
395	.setMessage(ACTIVITY_FAILED_MESSAGE)	1✔
396	.build();	1✔
397	completionCallback.apply(Optional.empty(), new FailureResult(failure, true));	1✔
398	}	1✔
399
400	private void notifyTimedOut() {
401	ActivityTaskTimedOutEventAttributes timedOut =	1✔
402	currentEvent.getActivityTaskTimedOutEventAttributes();	1✔
403
404	ActivityFailureInfo failureInfo =
405	ActivityFailureInfo.newBuilder()	1✔
406	.setActivityId(activityId)	1✔
407	.setActivityType(activityType)	1✔
408	.setRetryState(timedOut.getRetryState())	1✔
409	.setScheduledEventId(timedOut.getScheduledEventId())	1✔
410	.setStartedEventId(timedOut.getStartedEventId())	1✔
411	.build();	1✔
412	Failure failure =
413	Failure.newBuilder()	1✔
414	.setActivityFailureInfo(failureInfo)	1✔
415	.setCause(timedOut.getFailure())	1✔
416	.setMessage(ACTIVITY_TIMED_OUT_MESSAGE)	1✔
417	.build();	1✔
418	completionCallback.apply(Optional.empty(), new FailureResult(failure, true));	1✔
419	}	1✔
420
421	private void notifyCancellationFromEvent() {
422	if (cancellationType == ActivityCancellationType.WAIT_CANCELLATION_COMPLETED) {	1✔
423	ActivityTaskCanceledEventAttributes canceledAttr =	1✔
424	currentEvent.getActivityTaskCanceledEventAttributes();	1✔
425	Failure canceledFailure =
426	Failure.newBuilder()	1✔
427	.setSource(JAVA_SDK)	1✔
428	.setCanceledFailureInfo(	1✔
429	CanceledFailureInfo.newBuilder().setDetails(canceledAttr.getDetails()))	1✔
430	.build();	1✔
431
432	ActivityFailureInfo failureInfo =
433	ActivityFailureInfo.newBuilder()	1✔
434	.setActivityId(activityId)	1✔
435	.setActivityType(activityType)	1✔
436	.setScheduledEventId(canceledAttr.getScheduledEventId())	1✔
437	.setStartedEventId(canceledAttr.getStartedEventId())	1✔
438	.build();	1✔
439	Failure failure =
440	Failure.newBuilder()	1✔
441	.setActivityFailureInfo(failureInfo)	1✔
442	.setCause(canceledFailure)	1✔
443	.setMessage(ACTIVITY_CANCELED_MESSAGE)	1✔
444	.build();	1✔
445
446	completionCallback.apply(Optional.empty(), new FailureResult(failure, true));	1✔
447	}
448	}	1✔
449
450	private void createRequestCancelActivityTaskCommand() {
451	addCommand(	1✔
452	Command.newBuilder()	1✔
453	.setCommandType(CommandType.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK)	1✔
454	.setRequestCancelActivityTaskCommandAttributes(	1✔
455	RequestCancelActivityTaskCommandAttributes.newBuilder()	1✔
456	.setScheduledEventId(getInitialCommandEventId()))	1✔
457	.build());	1✔
458	parameters = null; // avoid retaining large input for the duration of the activity	1✔
459	}	1✔
460
461	public static class FailureResult {
462	private final @Nonnull Failure failure;
463	private final boolean fromEvent;
464
465	public FailureResult(@Nonnull Failure failure, boolean fromEvent) {	1✔
466	this.failure = failure;	1✔
467	this.fromEvent = fromEvent;	1✔
468	}	1✔
469
470	@Nonnull
471	public Failure getFailure() {
472	return failure;	1✔
473	}
474
475	/**
476	* @return true if this failure is created from the event during event <-> command matching
477	* phase.
478	*/
479	public boolean isFromEvent() {
480	return fromEvent;	1✔
481	}
482	}
483	}

temporalio / sdk-java / #159

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous