diff --git a/temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachine.java b/temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachine.java index b1605ad5d4..cfbc7e131f 100644 --- a/temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachine.java +++ b/temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachine.java @@ -199,6 +199,24 @@ StateMachine add( return this; } + /** Number of transitions taken so far; pass it to {@link #rollbackTo(int)}. */ + int transitionCount() { + return transitionHistory.size(); + } + + /** + * Undoes the transitions taken after {@code transitionCount} was read, restoring the state the + * machine had at that point. For when the request that caused those transitions is refused before + * its {@link RequestContext} is committed: the events the callbacks added are discarded with the + * context, and the in-memory state must not run ahead of history. + */ + void rollbackTo(int transitionCount) { + while (transitionHistory.size() > transitionCount) { + Transition undone = transitionHistory.remove(transitionHistory.size() - 1); + state = undone.from; + } + } + void action(Action action, RequestContext context, R request, long referenceId) { Transition transition = new Transition(state, action); @SuppressWarnings("unchecked") diff --git a/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableStateImpl.java b/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableStateImpl.java index 804cf7906c..43786adac9 100644 --- a/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableStateImpl.java +++ b/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableStateImpl.java @@ -53,6 +53,7 @@ import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; import java.util.function.LongSupplier; +import java.util.function.Predicate; import java.util.stream.Collectors; import java.util.stream.IntStream; import javax.annotation.Nonnull; @@ -451,7 +452,10 @@ public void completeWorkflowTask( List commands = request.getCommandsList(); List messages = new ArrayList<>(request.getMessagesList()); - completeWorkflowTaskUpdate( + AtomicReference refused = new AtomicReference<>(); + completeWorkflowTaskUpdateOrFailRefusedCommand( + request, + refused, ctx -> { if (ctx.getInitialEventId() != historySizeFromToken + 1) { throw Status.NOT_FOUND @@ -519,11 +523,18 @@ public void completeWorkflowTask( } long workflowTaskCompletedId = ctx.getNextEventId() - 1; + CompletionSnapshot beforeCompletion = new CompletionSnapshot(); + long completedTaskScheduledEventId = workflowTaskStateMachine.getData().scheduledEventId; try { workflowTaskStateMachine.action(StateMachines.Action.COMPLETE, ctx, request, 0); for (Command command : commands) { - processCommand( - ctx, command, messages, request.getIdentity(), workflowTaskCompletedId); + try { + processCommand( + ctx, command, messages, request.getIdentity(), workflowTaskCompletedId); + } catch (StatusRuntimeException e) { + refused.set(RefusedCommand.of(command, e, completedTaskScheduledEventId)); + throw e; + } } // Any messages not processed in processCommand need to be handled after all commands for (Message message : messages) { @@ -610,6 +621,13 @@ public void completeWorkflowTask( } data.queryBuffer.clear(); })); + } catch (RuntimeException e) { + // The exception discards this update, so the in-memory state must not run ahead of + // history. A workflow task left at NONE while history still says STARTED could never + // be failed, timed out, or redelivered (#3088), and a timer or activity added by a + // command before the refused one would reject the worker's replay as a duplicate. + beforeCompletion.restore(); + throw e; } finally { ctx.unlockTimer("completeWorkflowTask"); } @@ -617,6 +635,217 @@ public void completeWorkflowTask( request.hasStickyAttributes() ? request.getStickyAttributes() : null); } + /** + * Runs a workflow task completion update. If command processing refused a command with a caller + * error, the update was discarded and the workflow task rolled back to STARTED; record the + * refusal the way the real server does before surfacing the error to the worker. + */ + private void completeWorkflowTaskUpdateOrFailRefusedCommand( + RespondWorkflowTaskCompletedRequest request, + AtomicReference refused, + UpdateProcedure updater, + StickyExecutionAttributes attributes) { + try { + completeWorkflowTaskUpdate(updater, attributes); + } catch (StatusRuntimeException e) { + if (refused.get() == null) { + throw e; + } + failWorkflowTaskOnRefusedCommand(refused.get(), request); + throw e; + } + } + + /** + * Records the refusal of a command the way the real server does: fail the workflow task with the + * command's cause and schedule a new one (after the second attempt the failure is dropped and the + * task is left to time out), and surface the cause to the worker as INVALID_ARGUMENT. + */ + private void failWorkflowTaskOnRefusedCommand( + RefusedCommand refused, RespondWorkflowTaskCompletedRequest request) { + lock.lock(); + try { + // The task may have timed out, and a new one may have been scheduled and started, between + // the refused update and this one. Only the refused task is failed; the check runs before + // completeWorkflowTaskUpdate so a stale refusal does not touch the newer task's sticky + // settings either. + if (workflowTaskStateMachine.getState() != State.STARTED + || workflowTaskStateMachine.getData().scheduledEventId != refused.scheduledEventId) { + throw Status.INVALID_ARGUMENT.withDescription(refused.message).asRuntimeException(); + } + completeWorkflowTaskUpdate( + ctx -> { + failWorkflowTaskWithAReason( + refused.cause, new ServerFailure(refused.message, true), ctx, request, true); + ctx.setExceptionIfEmpty( + Status.INVALID_ARGUMENT.withDescription(refused.message).asRuntimeException()); + }, + null); + } finally { + lock.unlock(); + } + } + + /** + * The in-memory state a workflow task completion can change, captured before the completion is + * applied so a refused completion can be undone. The real server discards all mutable state + * changes of a failed workflow task; here the maps are restored and every state machine that + * existed is rolled back to the transitions it had, which also drops the machines the completion + * created. Callback side effects on state machine data are not undone. + */ + private class CompletionSnapshot { + private final Map> activities; + private final Map activityById; + private final Map> childWorkflows; + private final Map> nexusOperations; + private final Map nexusCancelRequestedEventIds; + private final Map> timers; + private final Map> externalSignals; + private final Map> externalCancellations; + private final Map> updates; + private final Map, Integer> transitionCounts = new IdentityHashMap<>(); + + CompletionSnapshot() { + activities = new HashMap<>(TestWorkflowMutableStateImpl.this.activities); + activityById = new HashMap<>(TestWorkflowMutableStateImpl.this.activityById); + childWorkflows = new HashMap<>(TestWorkflowMutableStateImpl.this.childWorkflows); + nexusOperations = new HashMap<>(TestWorkflowMutableStateImpl.this.nexusOperations); + nexusCancelRequestedEventIds = + new HashMap<>(TestWorkflowMutableStateImpl.this.nexusCancelRequestedEventIds); + timers = new HashMap<>(TestWorkflowMutableStateImpl.this.timers); + externalSignals = new HashMap<>(TestWorkflowMutableStateImpl.this.externalSignals); + externalCancellations = + new HashMap<>(TestWorkflowMutableStateImpl.this.externalCancellations); + updates = new HashMap<>(TestWorkflowMutableStateImpl.this.updates); + record(workflow); + record(workflowTaskStateMachine); + activities.values().forEach(this::record); + childWorkflows.values().forEach(this::record); + nexusOperations.values().forEach(this::record); + timers.values().forEach(this::record); + externalSignals.values().forEach(this::record); + externalCancellations.values().forEach(this::record); + updates.values().forEach(this::record); + } + + private void record(StateMachine machine) { + transitionCounts.put(machine, machine.transitionCount()); + } + + void restore() { + restore(TestWorkflowMutableStateImpl.this.activities, activities); + restore(TestWorkflowMutableStateImpl.this.activityById, activityById); + restore(TestWorkflowMutableStateImpl.this.childWorkflows, childWorkflows); + restore(TestWorkflowMutableStateImpl.this.nexusOperations, nexusOperations); + restore( + TestWorkflowMutableStateImpl.this.nexusCancelRequestedEventIds, + nexusCancelRequestedEventIds); + restore(TestWorkflowMutableStateImpl.this.timers, timers); + restore(TestWorkflowMutableStateImpl.this.externalSignals, externalSignals); + restore(TestWorkflowMutableStateImpl.this.externalCancellations, externalCancellations); + restore(TestWorkflowMutableStateImpl.this.updates, updates); + transitionCounts.forEach(StateMachine::rollbackTo); + } + + private void restore(Map live, Map saved) { + live.clear(); + live.putAll(saved); + } + } + + /** A command that command processing refused with a caller error. */ + private static class RefusedCommand { + final WorkflowTaskFailedCause cause; + final String message; + + /** Identifies the workflow task whose completion was refused. */ + final long scheduledEventId; + + private RefusedCommand(WorkflowTaskFailedCause cause, String message, long scheduledEventId) { + this.cause = cause; + this.message = message; + this.scheduledEventId = scheduledEventId; + } + + /** Null when the error is not a caller error or the command has no failure cause. */ + static RefusedCommand of(Command command, StatusRuntimeException e, long scheduledEventId) { + Status.Code code = e.getStatus().getCode(); + if (code != Status.Code.INVALID_ARGUMENT && code != Status.Code.FAILED_PRECONDITION) { + return null; + } + WorkflowTaskFailedCause cause = workflowTaskFailedCauseFor(command.getCommandType(), code); + if (cause == null) { + return null; + } + return new RefusedCommand( + cause, + ProtoEnumNameUtils.uniqueToSimplifiedName(cause) + ": " + e.getStatus().getDescription(), + scheduledEventId); + } + + /** + * The cause the real server records when it refuses a command of this type. The test server + * reports a duplicate timer or activity id as FAILED_PRECONDITION, which the real server + * records with its own cause. + */ + private static WorkflowTaskFailedCause workflowTaskFailedCauseFor( + CommandType type, Status.Code code) { + switch (type) { + case COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK: + return code == Status.Code.FAILED_PRECONDITION + ? WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_SCHEDULE_ACTIVITY_DUPLICATE_ID + : WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_SCHEDULE_ACTIVITY_ATTRIBUTES; + case COMMAND_TYPE_START_TIMER: + return code == Status.Code.FAILED_PRECONDITION + ? WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_START_TIMER_DUPLICATE_ID + : WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_START_TIMER_ATTRIBUTES; + case COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_REQUEST_CANCEL_ACTIVITY_ATTRIBUTES; + case COMMAND_TYPE_CANCEL_TIMER: + return WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_CANCEL_TIMER_ATTRIBUTES; + case COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_COMPLETE_WORKFLOW_EXECUTION_ATTRIBUTES; + case COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_FAIL_WORKFLOW_EXECUTION_ATTRIBUTES; + case COMMAND_TYPE_CANCEL_WORKFLOW_EXECUTION: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_CANCEL_WORKFLOW_EXECUTION_ATTRIBUTES; + case COMMAND_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_ATTRIBUTES; + case COMMAND_TYPE_RECORD_MARKER: + return WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_RECORD_MARKER_ATTRIBUTES; + case COMMAND_TYPE_PROTOCOL_MESSAGE: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_UPDATE_WORKFLOW_EXECUTION_MESSAGE; + case COMMAND_TYPE_CONTINUE_AS_NEW_WORKFLOW_EXECUTION: + return WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_CONTINUE_AS_NEW_ATTRIBUTES; + case COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_START_CHILD_EXECUTION_ATTRIBUTES; + case COMMAND_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_SIGNAL_WORKFLOW_EXECUTION_ATTRIBUTES; + case COMMAND_TYPE_UPSERT_WORKFLOW_SEARCH_ATTRIBUTES: + return WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_SEARCH_ATTRIBUTES; + case COMMAND_TYPE_MODIFY_WORKFLOW_PROPERTIES: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_MODIFY_WORKFLOW_PROPERTIES_ATTRIBUTES; + case COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_SCHEDULE_NEXUS_OPERATION_ATTRIBUTES; + case COMMAND_TYPE_REQUEST_CANCEL_NEXUS_OPERATION: + return WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_BAD_REQUEST_CANCEL_NEXUS_OPERATION_ATTRIBUTES; + default: + return null; + } + } + } + @Override public void applyOnConflictOptions(@Nonnull StartWorkflowExecutionRequest request) { update( @@ -640,6 +869,40 @@ public void applyOnConflictOptions(@Nonnull StartWorkflowExecutionRequest reques }); } + private boolean hasBufferedActivityFinish(long scheduledEventId) { + return hasBufferedEvent( + event -> { + switch (event.getEventType()) { + case EVENT_TYPE_ACTIVITY_TASK_COMPLETED: + return event.getActivityTaskCompletedEventAttributes().getScheduledEventId() + == scheduledEventId; + case EVENT_TYPE_ACTIVITY_TASK_FAILED: + return event.getActivityTaskFailedEventAttributes().getScheduledEventId() + == scheduledEventId; + case EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT: + return event.getActivityTaskTimedOutEventAttributes().getScheduledEventId() + == scheduledEventId; + case EVENT_TYPE_ACTIVITY_TASK_CANCELED: + return event.getActivityTaskCanceledEventAttributes().getScheduledEventId() + == scheduledEventId; + default: + return false; + } + }); + } + + /** Events that arrived while the current workflow task was in progress. */ + private boolean hasBufferedEvent(Predicate predicate) { + for (RequestContext buffered : workflowTaskStateMachine.getData().bufferedEvents) { + for (HistoryEvent event : buffered.getEvents()) { + if (predicate.test(event)) { + return true; + } + } + } + return false; + } + private void failWorkflowTaskWithAReason( WorkflowTaskFailedCause failedCause, ServerFailure eventAttributesFailure, @@ -930,25 +1193,31 @@ private void processRequestCancelExternalWorkflowExecution( StateMachines.newCancelExternalStateMachine(); externalCancellations.put(attr.getWorkflowId(), cancelStateMachine); cancelStateMachine.action(StateMachines.Action.INITIATE, ctx, attr, workflowTaskCompletedId); - ForkJoinPool.commonPool() - .execute( - () -> { - RequestCancelWorkflowExecutionRequest request = - RequestCancelWorkflowExecutionRequest.newBuilder() - .setWorkflowExecution( - WorkflowExecution.newBuilder().setWorkflowId(attr.getWorkflowId())) - .setNamespace(ctx.getNamespace()) - .setReason(attr.getReason()) - .build(); - CancelExternalWorkflowExecutionCallerInfo info = - new CancelExternalWorkflowExecutionCallerInfo( - ctx.getNamespace(), cancelStateMachine.getData().initiatedEventId, this); - try { - service.requestCancelWorkflowExecution(request, Optional.of(info)); - } catch (Exception e) { - log.error("Failure to request cancel external workflow", e); - } - }); + // Dispatched on commit so a completion that is refused by a later command cancels nothing. + ctx.onCommit( + (int historySize) -> + ForkJoinPool.commonPool() + .execute( + () -> { + RequestCancelWorkflowExecutionRequest request = + RequestCancelWorkflowExecutionRequest.newBuilder() + .setWorkflowExecution( + WorkflowExecution.newBuilder() + .setWorkflowId(attr.getWorkflowId())) + .setNamespace(ctx.getNamespace()) + .setReason(attr.getReason()) + .build(); + CancelExternalWorkflowExecutionCallerInfo info = + new CancelExternalWorkflowExecutionCallerInfo( + ctx.getNamespace(), + cancelStateMachine.getData().initiatedEventId, + this); + try { + service.requestCancelWorkflowExecution(request, Optional.of(info)); + } catch (Exception e) { + log.error("Failure to request cancel external workflow", e); + } + })); } @Override @@ -1019,6 +1288,20 @@ private void processRequestCancelActivityTask( long scheduledEventId = a.getScheduledEventId(); StateMachine activity = activities.get(scheduledEventId); if (activity == null) { + if (hasBufferedActivityFinish(scheduledEventId)) { + // The activity finished while this workflow task was running. Like the real server, + // record the cancel request and take no further action; the buffered finish event + // follows it in history. + ctx.addEvent( + HistoryEvent.newBuilder() + .setEventType(EventType.EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED) + .setActivityTaskCancelRequestedEventAttributes( + ActivityTaskCancelRequestedEventAttributes.newBuilder() + .setScheduledEventId(scheduledEventId) + .setWorkflowTaskCompletedEventId(workflowTaskCompletedId)) + .build()); + return; + } throw Status.FAILED_PRECONDITION .withDescription("ACTIVITY_UNKNOWN for scheduledEventId=" + scheduledEventId) .asRuntimeException(); @@ -1233,15 +1516,18 @@ private void processSignalExternalWorkflowExecution( StateMachines.newSignalExternalStateMachine(); externalSignals.put(signalId, signalStateMachine); signalStateMachine.action(StateMachines.Action.INITIATE, ctx, a, workflowTaskCompletedId); - ForkJoinPool.commonPool() - .execute( - () -> { - try { - service.signalExternalWorkflowExecution(signalId, a, this); - } catch (Exception e) { - log.error("Failure signalling an external workflow execution", e); - } - }); + // Dispatched on commit so a completion that is refused by a later command sends nothing. + ctx.onCommit( + (int historySize) -> + ForkJoinPool.commonPool() + .execute( + () -> { + try { + service.signalExternalWorkflowExecution(signalId, a, this); + } catch (Exception e) { + log.error("Failure signalling an external workflow execution", e); + } + })); ctx.lockTimer("processSignalExternalWorkflowExecution"); } @@ -1834,8 +2120,11 @@ private WorkflowTaskFailedCause processUpsertWorkflowSearchAttributes( RequestContext ctx, UpsertWorkflowSearchAttributesCommandAttributes attr, long workflowTaskCompletedId) { - visibilityStore.upsertSearchAttributesForExecution( - ctx.getExecutionId(), attr.getSearchAttributes()); + // Applied on commit so a completion that is refused by a later command changes nothing. + ctx.onCommit( + (int historySize) -> + visibilityStore.upsertSearchAttributesForExecution( + ctx.getExecutionId(), attr.getSearchAttributes())); UpsertWorkflowSearchAttributesEventAttributes.Builder upsertEventAttr = UpsertWorkflowSearchAttributesEventAttributes.newBuilder() @@ -1855,8 +2144,11 @@ private void processModifyWorkflowProperties( RequestContext ctx, ModifyWorkflowPropertiesCommandAttributes attr, long workflowTaskCompletedId) { - // Update workflow properties - currentMemo = mergeMemo(currentMemo, attr.getUpsertedMemo().getFieldsMap()); + // Merged on commit so a completion that is refused by a later command changes nothing, and + // in registration order so successive changes in one completion stack. + ctx.onCommit( + (int historySize) -> + currentMemo = mergeMemo(currentMemo, attr.getUpsertedMemo().getFieldsMap())); WorkflowPropertiesModifiedEventAttributes.Builder propModifiedEventAttr = WorkflowPropertiesModifiedEventAttributes.newBuilder() @@ -1890,7 +2182,13 @@ private WorkflowTaskFailedCause processProtocolMessageAttributes( messages.remove(msg); return msg; }) - .get(); + .orElseThrow( + () -> + Status.INVALID_ARGUMENT + .withDescription( + "ProtocolMessage command references unknown message id " + + attr.getMessageId()) + .asRuntimeException()); processMessage(ctx, orderedMsg, identity, workflowTaskCompletedId); return null; } diff --git a/temporal-test-server/src/test/java/io/temporal/testserver/functional/RefusedCommandTest.java b/temporal-test-server/src/test/java/io/temporal/testserver/functional/RefusedCommandTest.java new file mode 100644 index 0000000000..ad47cf2a71 --- /dev/null +++ b/temporal-test-server/src/test/java/io/temporal/testserver/functional/RefusedCommandTest.java @@ -0,0 +1,555 @@ +package io.temporal.testserver.functional; + +import static io.temporal.internal.common.InternalUtils.createNormalTaskQueue; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import com.google.protobuf.ByteString; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.temporal.api.command.v1.CancelTimerCommandAttributes; +import io.temporal.api.command.v1.Command; +import io.temporal.api.command.v1.CompleteWorkflowExecutionCommandAttributes; +import io.temporal.api.command.v1.ModifyWorkflowPropertiesCommandAttributes; +import io.temporal.api.command.v1.ProtocolMessageCommandAttributes; +import io.temporal.api.command.v1.RequestCancelActivityTaskCommandAttributes; +import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributes; +import io.temporal.api.command.v1.SignalExternalWorkflowExecutionCommandAttributes; +import io.temporal.api.command.v1.StartTimerCommandAttributes; +import io.temporal.api.command.v1.UpsertWorkflowSearchAttributesCommandAttributes; +import io.temporal.api.common.v1.ActivityType; +import io.temporal.api.common.v1.Memo; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.SearchAttributes; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.common.v1.WorkflowType; +import io.temporal.api.enums.v1.CommandType; +import io.temporal.api.enums.v1.EventType; +import io.temporal.api.enums.v1.WorkflowTaskFailedCause; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.taskqueue.v1.TaskQueue; +import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionResponse; +import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryRequest; +import io.temporal.api.workflowservice.v1.PollActivityTaskQueueRequest; +import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse; +import io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueRequest; +import io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueResponse; +import io.temporal.api.workflowservice.v1.RespondActivityTaskCompletedRequest; +import io.temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedRequest; +import io.temporal.api.workflowservice.v1.StartWorkflowExecutionRequest; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.internal.common.ProtobufTimeUtils; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import io.temporal.testing.internal.TestServiceUtils; +import io.temporal.testserver.TestServer; +import java.time.Duration; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +/** + * A workflow task completion that cancels a timer or an activity the server does not know about + * must fail the workflow task and schedule a new one, the way the real server does. Before the fix + * the test server refused the completion after it had already moved the workflow task state machine + * to NONE, so the task was never timed out or redelivered and the run hung forever. + */ +public class RefusedCommandTest { + + private static final String NAMESPACE = "namespace"; + private static final String TASK_QUEUE = "taskQueue"; + private static final String WORKFLOW_TYPE = "wfType"; + + private TestServer.InProcessTestServer testServer; + private WorkflowServiceStubs workflowServiceStubs; + + @Before + public void setUp() { + this.testServer = TestServer.createServer(true); + this.workflowServiceStubs = + WorkflowServiceStubs.newServiceStubs( + WorkflowServiceStubsOptions.newBuilder() + .setChannel(testServer.getChannel()) + .validateAndBuildWithDefaults()); + } + + @After + public void tearDown() { + this.workflowServiceStubs.shutdownNow(); + this.workflowServiceStubs.awaitTermination(1, TimeUnit.SECONDS); + this.testServer.close(); + } + + @Test + public void cancelTimerForUnknownTimerFailsWorkflowTaskAndReschedules() throws Exception { + PollWorkflowTaskQueueResponse task = startWorkflowAndPollFirstTask(); + + Command cancelUnknownTimer = + Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_CANCEL_TIMER) + .setCancelTimerCommandAttributes( + CancelTimerCommandAttributes.newBuilder().setTimerId("no-such-timer")) + .build(); + + StatusRuntimeException e = + assertThrows( + StatusRuntimeException.class, + () -> respondWorkflowTaskCompleted(task.getTaskToken(), cancelUnknownTimer)); + assertEquals(Status.Code.INVALID_ARGUMENT, e.getStatus().getCode()); + + List history = getHistory(task); + assertWorkflowTaskFailedAndRescheduled( + history, WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_CANCEL_TIMER_ATTRIBUTES); + assertFalse( + "no timer was canceled", + history.stream().anyMatch(ev -> ev.getEventType() == EventType.EVENT_TYPE_TIMER_CANCELED)); + + assertWorkflowTaskRedeliveredAndCompletes(task); + } + + @Test + public void requestCancelActivityForUnknownActivityFailsWorkflowTaskAndReschedules() + throws Exception { + PollWorkflowTaskQueueResponse task = startWorkflowAndPollFirstTask(); + + Command cancelUnknownActivity = + Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK) + .setRequestCancelActivityTaskCommandAttributes( + RequestCancelActivityTaskCommandAttributes.newBuilder().setScheduledEventId(12345)) + .build(); + + StatusRuntimeException e = + assertThrows( + StatusRuntimeException.class, + () -> respondWorkflowTaskCompleted(task.getTaskToken(), cancelUnknownActivity)); + assertEquals(Status.Code.INVALID_ARGUMENT, e.getStatus().getCode()); + + assertWorkflowTaskFailedAndRescheduled( + getHistory(task), + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_REQUEST_CANCEL_ACTIVITY_ATTRIBUTES); + + assertWorkflowTaskRedeliveredAndCompletes(task); + } + + @Test + public void scheduleActivityWithoutTaskQueueFailsWorkflowTaskAndReschedules() throws Exception { + PollWorkflowTaskQueueResponse task = startWorkflowAndPollFirstTask(); + + Command scheduleWithoutTaskQueue = + Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK) + .setScheduleActivityTaskCommandAttributes( + ScheduleActivityTaskCommandAttributes.newBuilder() + .setActivityId("activity") + .setActivityType(ActivityType.newBuilder().setName("activity")) + .setScheduleToCloseTimeout( + ProtobufTimeUtils.toProtoDuration(Duration.ofSeconds(60)))) + .build(); + + StatusRuntimeException e = + assertThrows( + StatusRuntimeException.class, + () -> respondWorkflowTaskCompleted(task.getTaskToken(), scheduleWithoutTaskQueue)); + assertEquals(Status.Code.INVALID_ARGUMENT, e.getStatus().getCode()); + + assertWorkflowTaskFailedAndRescheduled( + getHistory(task), + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_SCHEDULE_ACTIVITY_ATTRIBUTES); + + assertWorkflowTaskRedeliveredAndCompletes(task); + } + + @Test + public void startTimerWithDuplicateIdFailsWorkflowTaskAndReschedules() throws Exception { + PollWorkflowTaskQueueResponse firstTask = startWorkflowAndPollFirstTask(); + respondWorkflowTaskCompleted(firstTask.getTaskToken(), startTimerCommand("timer")); + + TestServiceUtils.signalWorkflow( + firstTask.getWorkflowExecution(), NAMESPACE, workflowServiceStubs); + PollWorkflowTaskQueueResponse secondTask = + TestServiceUtils.pollWorkflowTaskQueue( + NAMESPACE, createNormalTaskQueue(TASK_QUEUE), workflowServiceStubs); + + StatusRuntimeException e = + assertThrows( + StatusRuntimeException.class, + () -> + respondWorkflowTaskCompleted( + secondTask.getTaskToken(), startTimerCommand("timer"))); + assertEquals(Status.Code.INVALID_ARGUMENT, e.getStatus().getCode()); + + assertWorkflowTaskFailedAndRescheduled( + getHistory(firstTask), + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_START_TIMER_DUPLICATE_ID); + + assertWorkflowTaskRedeliveredAndCompletes(firstTask); + } + + @Test + public void commandsBeforeTheRefusedOneAreUndoneSoTheReplayIsAccepted() throws Exception { + PollWorkflowTaskQueueResponse task = startWorkflowAndPollFirstTask(); + + Command cancelUnknownTimer = + Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_CANCEL_TIMER) + .setCancelTimerCommandAttributes( + CancelTimerCommandAttributes.newBuilder().setTimerId("no-such-timer")) + .build(); + assertThrows( + StatusRuntimeException.class, + () -> + respondWorkflowTaskCompleted( + task.getTaskToken(), startTimerCommand("timer"), cancelUnknownTimer)); + assertWorkflowTaskFailedAndRescheduled( + getHistory(task), + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_CANCEL_TIMER_ATTRIBUTES); + + // The worker replays from history, which has no timer, so it sends StartTimer again. The + // timer the refused completion added must be gone or this is rejected as a duplicate. + PollWorkflowTaskQueueResponse redelivered = pollWorkflowTask(); + respondWorkflowTaskCompleted(redelivered.getTaskToken(), startTimerCommand("timer")); + assertTrue( + "expected the replayed StartTimer to be accepted, history: " + eventTypes(getHistory(task)), + getHistory(task).stream() + .anyMatch(ev -> ev.getEventType() == EventType.EVENT_TYPE_TIMER_STARTED)); + } + + @Test + public void requestCancelActivityThatFinishedDuringTheTaskIsAccepted() throws Exception { + PollWorkflowTaskQueueResponse firstTask = startWorkflowAndPollFirstTask(); + respondWorkflowTaskCompleted(firstTask.getTaskToken(), scheduleActivityTaskCommand()); + + // A signal schedules a second workflow task; poll it so it is in flight. + TestServiceUtils.signalWorkflow( + firstTask.getWorkflowExecution(), NAMESPACE, workflowServiceStubs); + PollWorkflowTaskQueueResponse secondTask = + TestServiceUtils.pollWorkflowTaskQueue( + NAMESPACE, createNormalTaskQueue(TASK_QUEUE), workflowServiceStubs); + + // Complete the activity while the second workflow task is running, so its completion event + // is buffered and the activity no longer has a state machine. + PollActivityTaskQueueResponse activityTask = + workflowServiceStubs + .blockingStub() + .pollActivityTaskQueue( + PollActivityTaskQueueRequest.newBuilder() + .setNamespace(NAMESPACE) + .setTaskQueue(createNormalTaskQueue(TASK_QUEUE)) + .build()); + long scheduledEventId = + getHistory(firstTask).stream() + .filter(ev -> ev.getEventType() == EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED) + .findFirst() + .orElseThrow(() -> new AssertionError("activity was not scheduled")) + .getEventId(); + workflowServiceStubs + .blockingStub() + .respondActivityTaskCompleted( + RespondActivityTaskCompletedRequest.newBuilder() + .setTaskToken(activityTask.getTaskToken()) + .build()); + + // The worker, which has not seen the completion yet, cancels the activity. + Command cancelActivity = + Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK) + .setRequestCancelActivityTaskCommandAttributes( + RequestCancelActivityTaskCommandAttributes.newBuilder() + .setScheduledEventId(scheduledEventId)) + .build(); + respondWorkflowTaskCompleted(secondTask.getTaskToken(), cancelActivity); + + List events = eventTypes(getHistory(firstTask)); + assertFalse( + "the cancel must not fail the workflow task, history: " + events, + events.contains(EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED)); + int cancelRequested = events.indexOf(EventType.EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED); + int completed = events.indexOf(EventType.EVENT_TYPE_ACTIVITY_TASK_COMPLETED); + assertTrue("expected ActivityTaskCancelRequested, history: " + events, cancelRequested >= 0); + assertTrue( + "the buffered completion must follow the cancel request, history: " + events, + completed > cancelRequested); + } + + @Test + public void searchAttributesAndMemoOfARefusedCompletionAreNotApplied() throws Exception { + PollWorkflowTaskQueueResponse task = startWorkflowAndPollFirstTask(); + Payload one = DefaultDataConverter.newDefaultInstance().toPayload(1).get(); + Command upsert = + Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_UPSERT_WORKFLOW_SEARCH_ATTRIBUTES) + .setUpsertWorkflowSearchAttributesCommandAttributes( + UpsertWorkflowSearchAttributesCommandAttributes.newBuilder() + .setSearchAttributes( + SearchAttributes.newBuilder().putIndexedFields("CustomIntField", one))) + .build(); + Command memo = + Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_MODIFY_WORKFLOW_PROPERTIES) + .setModifyWorkflowPropertiesCommandAttributes( + ModifyWorkflowPropertiesCommandAttributes.newBuilder() + .setUpsertedMemo(Memo.newBuilder().putFields("memoKey", one))) + .build(); + + assertThrows( + StatusRuntimeException.class, + () -> + respondWorkflowTaskCompleted( + task.getTaskToken(), upsert, memo, cancelTimerCommand("no-such-timer"))); + assertWorkflowTaskFailedAndRescheduled( + getHistory(task), + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_CANCEL_TIMER_ATTRIBUTES); + + // Neither effect of the refused completion is visible: not in history, not in Describe. + List types = eventTypes(getHistory(task)); + assertFalse(types.contains(EventType.EVENT_TYPE_UPSERT_WORKFLOW_SEARCH_ATTRIBUTES)); + assertFalse(types.contains(EventType.EVENT_TYPE_WORKFLOW_PROPERTIES_MODIFIED)); + DescribeWorkflowExecutionResponse described = describe(task.getWorkflowExecution()); + assertFalse( + described + .getWorkflowExecutionInfo() + .getSearchAttributes() + .containsIndexedFields("CustomIntField")); + assertFalse(described.getWorkflowExecutionInfo().getMemo().containsFields("memoKey")); + + // The same changes on the redelivered task are applied. + PollWorkflowTaskQueueResponse redelivered = pollWorkflowTask(); + respondWorkflowTaskCompleted(redelivered.getTaskToken(), upsert, memo); + described = describe(task.getWorkflowExecution()); + // The store adds type metadata to the stored value, so compare the data only. + assertEquals( + one.getData(), + described + .getWorkflowExecutionInfo() + .getSearchAttributes() + .getIndexedFieldsOrThrow("CustomIntField") + .getData()); + assertEquals(one, described.getWorkflowExecutionInfo().getMemo().getFieldsOrThrow("memoKey")); + } + + @Test + public void aRefusedCompletionDoesNotSignalTheExternalWorkflow() throws Exception { + PollWorkflowTaskQueueResponse task = startWorkflowAndPollFirstTask(); + String targetId = "target-" + UUID.randomUUID(); + workflowServiceStubs + .blockingStub() + .startWorkflowExecution( + StartWorkflowExecutionRequest.newBuilder() + .setNamespace(NAMESPACE) + .setRequestId(UUID.randomUUID().toString()) + .setWorkflowId(targetId) + .setWorkflowType(WorkflowType.newBuilder().setName(WORKFLOW_TYPE)) + .setTaskQueue(TaskQueue.newBuilder().setName("target-" + TASK_QUEUE)) + .setWorkflowRunTimeout(ProtobufTimeUtils.toProtoDuration(Duration.ofSeconds(100))) + .setWorkflowTaskTimeout(ProtobufTimeUtils.toProtoDuration(Duration.ofSeconds(100))) + .build()); + WorkflowExecution target = WorkflowExecution.newBuilder().setWorkflowId(targetId).build(); + Command signal = + Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION) + .setSignalExternalWorkflowExecutionCommandAttributes( + SignalExternalWorkflowExecutionCommandAttributes.newBuilder() + .setExecution(target) + .setSignalName("signal")) + .build(); + + assertThrows( + StatusRuntimeException.class, + () -> + respondWorkflowTaskCompleted( + task.getTaskToken(), signal, cancelTimerCommand("no-such-timer"))); + assertWorkflowTaskFailedAndRescheduled( + getHistory(task), + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_CANCEL_TIMER_ATTRIBUTES); + + // The signal was dispatched on commit, and the refused completion never committed. Delivery is + // asynchronous, so give a wrongly dispatched signal time to land before checking. + Thread.sleep(500); + assertFalse( + eventTypes(getHistory(task)) + .contains(EventType.EVENT_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_INITIATED)); + assertFalse( + eventTypes(getHistory(target)).contains(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED)); + + // The same signal on the redelivered task is delivered. + PollWorkflowTaskQueueResponse redelivered = pollWorkflowTask(); + respondWorkflowTaskCompleted(redelivered.getTaskToken(), signal); + long deadline = System.currentTimeMillis() + 5_000; + while (!eventTypes(getHistory(target)) + .contains(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED) + && System.currentTimeMillis() < deadline) { + Thread.sleep(50); + } + assertTrue( + "expected the target to be signaled, history: " + eventTypes(getHistory(target)), + eventTypes(getHistory(target)).contains(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED)); + } + + @Test + public void aProtocolMessageForAnUnknownMessageFailsTheTaskWithTheUpdateCause() throws Exception { + PollWorkflowTaskQueueResponse task = startWorkflowAndPollFirstTask(); + Command message = + Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_PROTOCOL_MESSAGE) + .setProtocolMessageCommandAttributes( + ProtocolMessageCommandAttributes.newBuilder().setMessageId("no-such-message")) + .build(); + + StatusRuntimeException e = + assertThrows( + StatusRuntimeException.class, + () -> respondWorkflowTaskCompleted(task.getTaskToken(), message)); + assertEquals(Status.Code.INVALID_ARGUMENT, e.getStatus().getCode()); + assertWorkflowTaskFailedAndRescheduled( + getHistory(task), + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_BAD_UPDATE_WORKFLOW_EXECUTION_MESSAGE); + assertWorkflowTaskRedeliveredAndCompletes(task); + } + + private Command cancelTimerCommand(String timerId) { + return Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_CANCEL_TIMER) + .setCancelTimerCommandAttributes( + CancelTimerCommandAttributes.newBuilder().setTimerId(timerId)) + .build(); + } + + private DescribeWorkflowExecutionResponse describe(WorkflowExecution execution) { + return workflowServiceStubs + .blockingStub() + .describeWorkflowExecution( + DescribeWorkflowExecutionRequest.newBuilder() + .setNamespace(NAMESPACE) + .setExecution(execution) + .build()); + } + + private List getHistory(WorkflowExecution execution) { + return workflowServiceStubs + .blockingStub() + .getWorkflowExecutionHistory( + GetWorkflowExecutionHistoryRequest.newBuilder() + .setNamespace(NAMESPACE) + .setExecution(execution) + .build()) + .getHistory() + .getEventsList(); + } + + private Command startTimerCommand(String timerId) { + return Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_START_TIMER) + .setStartTimerCommandAttributes( + StartTimerCommandAttributes.newBuilder() + .setTimerId(timerId) + .setStartToFireTimeout(ProtobufTimeUtils.toProtoDuration(Duration.ofHours(1)))) + .build(); + } + + private Command scheduleActivityTaskCommand() { + return Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK) + .setScheduleActivityTaskCommandAttributes( + ScheduleActivityTaskCommandAttributes.newBuilder() + .setActivityId("activity") + .setActivityType(ActivityType.newBuilder().setName("activity")) + .setTaskQueue(TaskQueue.newBuilder().setName(TASK_QUEUE)) + .setScheduleToCloseTimeout( + ProtobufTimeUtils.toProtoDuration(Duration.ofSeconds(60)))) + .build(); + } + + private PollWorkflowTaskQueueResponse startWorkflowAndPollFirstTask() throws Exception { + TestServiceUtils.startWorkflowExecution( + NAMESPACE, TASK_QUEUE, WORKFLOW_TYPE, workflowServiceStubs); + return TestServiceUtils.pollWorkflowTaskQueue( + NAMESPACE, createNormalTaskQueue(TASK_QUEUE), workflowServiceStubs); + } + + private void assertWorkflowTaskFailedAndRescheduled( + List history, WorkflowTaskFailedCause expectedCause) { + int failedIndex = -1; + for (int i = 0; i < history.size(); i++) { + if (history.get(i).getEventType() == EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED) { + failedIndex = i; + break; + } + } + assertTrue( + "expected a WorkflowTaskFailed event, history: " + eventTypes(history), failedIndex >= 0); + assertEquals( + expectedCause, history.get(failedIndex).getWorkflowTaskFailedEventAttributes().getCause()); + assertTrue( + "expected a new WorkflowTaskScheduled after the failure, history: " + eventTypes(history), + history.subList(failedIndex + 1, history.size()).stream() + .anyMatch(ev -> ev.getEventType() == EventType.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED)); + } + + /** The failed task must be redelivered, and the run must still be able to complete. */ + private PollWorkflowTaskQueueResponse pollWorkflowTask() { + PollWorkflowTaskQueueResponse task = + workflowServiceStubs + .blockingStub() + .withDeadlineAfter(10, TimeUnit.SECONDS) + .pollWorkflowTaskQueue( + PollWorkflowTaskQueueRequest.newBuilder() + .setNamespace(NAMESPACE) + .setTaskQueue(createNormalTaskQueue(TASK_QUEUE)) + .build()); + assertFalse("expected a workflow task to be delivered", task.getTaskToken().isEmpty()); + return task; + } + + private void assertWorkflowTaskRedeliveredAndCompletes(PollWorkflowTaskQueueResponse firstTask) + throws Exception { + PollWorkflowTaskQueueResponse redelivered = pollWorkflowTask(); + + Command complete = + Command.newBuilder() + .setCommandType(CommandType.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION) + .setCompleteWorkflowExecutionCommandAttributes( + CompleteWorkflowExecutionCommandAttributes.getDefaultInstance()) + .build(); + respondWorkflowTaskCompleted(redelivered.getTaskToken(), complete); + + assertTrue( + "expected the workflow to complete after the redelivered task", + getHistory(firstTask).stream() + .anyMatch( + ev -> ev.getEventType() == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED)); + } + + private void respondWorkflowTaskCompleted(ByteString taskToken, Command... commands) { + RespondWorkflowTaskCompletedRequest.Builder request = + RespondWorkflowTaskCompletedRequest.newBuilder().setTaskToken(taskToken); + for (Command command : commands) { + request.addCommands(command); + } + workflowServiceStubs.blockingStub().respondWorkflowTaskCompleted(request.build()); + } + + private List getHistory(PollWorkflowTaskQueueResponse task) { + return workflowServiceStubs + .blockingStub() + .getWorkflowExecutionHistory( + GetWorkflowExecutionHistoryRequest.newBuilder() + .setNamespace(NAMESPACE) + .setExecution(task.getWorkflowExecution()) + .build()) + .getHistory() + .getEventsList(); + } + + private static List eventTypes(List history) { + return history.stream() + .map(HistoryEvent::getEventType) + .collect(java.util.stream.Collectors.toList()); + } +}