diff --git a/tests/xdc/buffered_events_replication_helpers_test.go b/tests/xdc/buffered_events_replication_helpers_test.go new file mode 100644 index 0000000000..bf69cbd21f --- /dev/null +++ b/tests/xdc/buffered_events_replication_helpers_test.go @@ -0,0 +1,1325 @@ +package xdc + +import ( + "context" + "errors" + "time" + + "github.com/google/uuid" + "github.com/nexus-rpc/sdk-go/nexus" + "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + failurepb "go.temporal.io/api/failure/v1" + historypb "go.temporal.io/api/history/v1" + nexuspb "go.temporal.io/api/nexus/v1" + "go.temporal.io/api/operatorservice/v1" + protocolpb "go.temporal.io/api/protocol/v1" + replicationpb "go.temporal.io/api/replication/v1" + "go.temporal.io/api/serviceerror" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + updatepb "go.temporal.io/api/update/v1" + workflowpb "go.temporal.io/api/workflow/v1" + "go.temporal.io/api/workflowservice/v1" + sdkclient "go.temporal.io/sdk/client" + "go.temporal.io/server/api/adminservice/v1" + "go.temporal.io/server/api/historyservice/v1" + replicationspb "go.temporal.io/server/api/replication/v1" + "go.temporal.io/server/chasm" + "go.temporal.io/server/common" + "go.temporal.io/server/common/dynamicconfig" + commonnexus "go.temporal.io/server/common/nexus" + "go.temporal.io/server/common/nexus/nexusrpc" + "go.temporal.io/server/common/nexus/nexustest" + "go.temporal.io/server/common/payloads" + "go.temporal.io/server/common/persistence" + serviceerrors "go.temporal.io/server/common/serviceerror" + "go.temporal.io/server/common/testing/await" + "go.temporal.io/server/common/testing/protoutils" + "go.temporal.io/server/common/testing/testhooks" + "go.temporal.io/server/components/nexusoperations" + "go.temporal.io/server/tests/testcore" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/durationpb" +) + +type blockedReplicationTask struct { + execute func() error + result chan error +} + +type bufferedEventsNamespace struct { + Name string + ID string +} + +type startBufferedEventsWorkflowArgs struct { + Namespace string + WorkflowID string + TaskQueue *taskqueuepb.TaskQueue +} + +type workflowTaskCompletion struct { + Task *workflowservice.PollWorkflowTaskQueueResponse + Commands []*commandpb.Command +} + +type bufferedEventExpectation struct { + Namespace string + Execution *commonpb.WorkflowExecution + EventTypes []enumspb.EventType +} + +type workflowSignal struct { + Namespace string + Execution *commonpb.WorkflowExecution + SignalName string +} + +type activeClusterSignal struct { + Namespace bufferedEventsNamespace + Execution *commonpb.WorkflowExecution + SignalName string +} + +type workflowCancellation struct { + Namespace string + Execution *commonpb.WorkflowExecution +} + +type workflowTermination struct { + Namespace string + Execution *commonpb.WorkflowExecution + Reason string +} + +type naturallyBufferedConflict struct { + Namespace bufferedEventsNamespace + Execution *commonpb.WorkflowExecution + ReplicationToOldActive <-chan *blockedReplicationTask + ReplicationToNewActive <-chan *blockedReplicationTask + ExpectedEventTypes []enumspb.EventType + WinnerSignal string +} + +type bufferedInputsExpectation struct { + Namespace bufferedEventsNamespace + Execution *commonpb.WorkflowExecution + UpdateID string + OptionsRequestID string + EventTypes []enumspb.EventType +} + +type bufferedNexusCallback struct { + operation string + url string + token string +} + +type bufferedNexusOperationCompletion struct { + Callback bufferedNexusCallback + Result string +} + +type bufferedNexusLosingBranch struct { + Namespace bufferedEventsNamespace + Execution *commonpb.WorkflowExecution + ScheduledEventID int64 +} + +func (s *xdcBaseSuite) createBufferedEventsNamespace(ctx context.Context) bufferedEventsNamespace { + name := s.createGlobalNamespace() + response, err := s.clusters[0].FrontendClient().DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{ + Namespace: name, + }) + s.Require().NoError(err) + return bufferedEventsNamespace{Name: name, ID: response.NamespaceInfo.Id} +} + +func (s *xdcBaseSuite) enableWorkflowPauseForTest() { + s.T().Helper() + for _, cluster := range s.clusters { + cluster.OverrideDynamicConfig(s.T(), dynamicconfig.WorkflowPauseEnabled, true) + } +} + +func (s *xdcBaseSuite) signalWorkflow(ctx context.Context, signal workflowSignal) { + s.T().Helper() + _, err := s.clusters[0].FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: signal.Namespace, + WorkflowExecution: signal.Execution, + SignalName: signal.SignalName, + RequestId: uuid.NewString(), + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) +} + +func (s *xdcBaseSuite) requestWorkflowCancellationEventually(ctx context.Context, target workflowCancellation) { + s.T().Helper() + await.Require(ctx, s.T(), func(t *await.T) { + _, err := s.clusters[0].FrontendClient().DescribeWorkflowExecution(t.Context(), &workflowservice.DescribeWorkflowExecutionRequest{ + Namespace: target.Namespace, + Execution: target.Execution, + }) + require.NoError(t, err) + }, replicationWaitTime, replicationCheckInterval) + _, err := s.clusters[0].FrontendClient().RequestCancelWorkflowExecution(ctx, &workflowservice.RequestCancelWorkflowExecutionRequest{ + Namespace: target.Namespace, + WorkflowExecution: target.Execution, + RequestId: uuid.NewString(), + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) +} + +func (s *xdcBaseSuite) terminateWorkflow(ctx context.Context, target workflowTermination) { + s.T().Helper() + _, err := s.clusters[0].FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{ + Namespace: target.Namespace, + WorkflowExecution: target.Execution, + Reason: target.Reason, + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) +} + +func (s *xdcBaseSuite) terminateWorkflowEventually(ctx context.Context, target workflowTermination) { + s.T().Helper() + await.Require(ctx, s.T(), func(t *await.T) { + _, err := s.clusters[0].FrontendClient().TerminateWorkflowExecution(t.Context(), &workflowservice.TerminateWorkflowExecutionRequest{ + Namespace: target.Namespace, + WorkflowExecution: target.Execution, + Reason: target.Reason, + Identity: "buffered-events-xdc-test", + }) + require.NoError(t, err) + }, replicationWaitTime, replicationCheckInterval) +} + +// The buffered-event conflict tests intentionally use a fixed two-cluster topology: +// +// 1. Cluster 0 starts active, establishes the common history, and holds the workflow task while +// naturally produced events enter its buffer. +// 2. Replication is blocked in both directions for this workflow. A namespace failover makes +// cluster 1 active, where a signal creates the winning branch. +// 3. Cluster 1 -> cluster 0 replication is released first. Cluster 0 is now passive; applying the +// winner resolves the conflict, failover-closes its outstanding workflow task, and flushes the +// buffer onto the losing branch. +// 4. Cluster 0 -> cluster 1 replication is then released so conflict resolution sees the losing +// branch and reapplies or skips its events against the winner. +// 5. Cluster 1 -> cluster 0 replication is finally released until both current histories converge. +// +// Replication channel names below identify the receiving cluster: replicationToOldActive contains +// tasks executing on cluster 0, and replicationToNewActive contains tasks executing on cluster 1. +func (s *xdcBaseSuite) finishNaturallyBufferedConflict( + ctx context.Context, + conflict naturallyBufferedConflict, +) []*historypb.HistoryEvent { + losingBranchMarker := "losing-branch-marker-" + uuid.NewString() + _, err := s.clusters[0].FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: conflict.Namespace.Name, + WorkflowExecution: conflict.Execution, + SignalName: losingBranchMarker, + RequestId: uuid.NewString(), + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) + s.Require().True(s.hasBufferedEventType(ctx, 0, conflict.Namespace.Name, conflict.Execution, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED)) + + s.failoverToNewActiveCluster(ctx, conflict.Namespace.Name) + s.writeSignalOnNewActive(ctx, activeClusterSignal{ + Namespace: conflict.Namespace, + Execution: conflict.Execution, + SignalName: conflict.WinnerSignal, + }) + s.releaseReplicationTask(ctx, conflict.ReplicationToOldActive) + s.assertNoBufferedEvents(ctx, 0, conflict.Namespace.Name, conflict.Execution) + losingHistory := s.findNonCurrentHistoryBranch(ctx, conflict.Namespace.Name, conflict.Namespace.ID, conflict.Execution, func(history []*historypb.HistoryEvent) bool { + for _, eventType := range conflict.ExpectedEventTypes { + if findHistoryEvent(history, eventType, nil) == nil { + return false + } + } + return true + }) + for _, eventType := range conflict.ExpectedEventTypes { + event := findHistoryEvent(losingHistory, eventType, nil) + s.Require().NotNil(event, "%s must be written to the losing branch", eventType) + s.Require().Positive(event.EventId) + s.Require().NotEqual(common.BufferedEventID, event.EventId) + } + await.Require(ctx, s.T(), func(t *await.T) { + s.tryReleaseReplicationTask(conflict.ReplicationToNewActive) + require.True( + t, + hasSignalNamed(s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, conflict.Namespace.Name, conflict.Execution), losingBranchMarker), + "the losing branch marker must be reapplied before checking the rest of the batch", + ) + }, replicationWaitTime, replicationCheckInterval) + for attempt := 0; attempt < 10 && !s.bufferedEventsHistoriesEqual(ctx, conflict.Namespace.Name, conflict.Execution); attempt++ { + s.releaseReplicationTask(ctx, conflict.ReplicationToOldActive) + } + return losingHistory +} + +func signalExternalWorkflowCommand(workflowID, runID, signalName string) *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_SignalExternalWorkflowExecutionCommandAttributes{ + SignalExternalWorkflowExecutionCommandAttributes: &commandpb.SignalExternalWorkflowExecutionCommandAttributes{ + Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID}, + SignalName: signalName, + }, + }, + } +} + +func cancelExternalWorkflowCommand(workflowID, runID string) *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_RequestCancelExternalWorkflowExecutionCommandAttributes{ + RequestCancelExternalWorkflowExecutionCommandAttributes: &commandpb.RequestCancelExternalWorkflowExecutionCommandAttributes{ + WorkflowId: workflowID, + RunId: runID, + }, + }, + } +} + +func startChildWorkflowCommand(childID string, taskQueue *taskqueuepb.TaskQueue, runTimeout time.Duration) *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_StartChildWorkflowExecutionCommandAttributes{ + StartChildWorkflowExecutionCommandAttributes: &commandpb.StartChildWorkflowExecutionCommandAttributes{ + WorkflowId: childID, + WorkflowType: &commonpb.WorkflowType{Name: "buffered-child"}, + TaskQueue: taskQueue, + WorkflowRunTimeout: durationpb.New(runTimeout), + WorkflowTaskTimeout: durationpb.New(30 * time.Second), + }, + }, + } +} + +func (s *xdcBaseSuite) startBufferedEventsWorkflow( + ctx context.Context, + args startBufferedEventsWorkflowArgs, +) *commonpb.WorkflowExecution { + response, err := s.clusters[0].FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{ + Namespace: args.Namespace, + WorkflowId: args.WorkflowID, + WorkflowType: &commonpb.WorkflowType{Name: "buffered-events-xdc"}, + TaskQueue: args.TaskQueue, + RequestId: uuid.NewString(), + WorkflowRunTimeout: durationpb.New(time.Minute), + WorkflowTaskTimeout: durationpb.New(2 * time.Minute), + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) + return &commonpb.WorkflowExecution{WorkflowId: args.WorkflowID, RunId: response.RunId} +} + +func scheduleActivityCommand(activityID string, taskQueue *taskqueuepb.TaskQueue, startToClose time.Duration) *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK, + Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{ + ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{ + ActivityId: activityID, + ActivityType: &commonpb.ActivityType{Name: activityID}, + TaskQueue: taskQueue, + ScheduleToCloseTimeout: durationpb.New(startToClose), + StartToCloseTimeout: durationpb.New(startToClose), + }, + }, + } +} + +func requestCancelActivityCommand(scheduledEventID int64) *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_REQUEST_CANCEL_ACTIVITY_TASK, + Attributes: &commandpb.Command_RequestCancelActivityTaskCommandAttributes{ + RequestCancelActivityTaskCommandAttributes: &commandpb.RequestCancelActivityTaskCommandAttributes{ + ScheduledEventId: scheduledEventID, + }, + }, + } +} + +func completeWorkflowCommand() *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{ + CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{}, + }, + } +} + +func failWorkflowCommand(message string) *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_FailWorkflowExecutionCommandAttributes{ + FailWorkflowExecutionCommandAttributes: &commandpb.FailWorkflowExecutionCommandAttributes{ + Failure: &failurepb.Failure{ + Message: message, + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{NonRetryable: true}, + }, + }, + }, + }, + } +} + +func cancelWorkflowCommand() *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_CANCEL_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_CancelWorkflowExecutionCommandAttributes{ + CancelWorkflowExecutionCommandAttributes: &commandpb.CancelWorkflowExecutionCommandAttributes{}, + }, + } +} + +func (s *xdcBaseSuite) pollBufferedActivityTask( + ctx context.Context, + ns string, + taskQueue *taskqueuepb.TaskQueue, +) *workflowservice.PollActivityTaskQueueResponse { + response, err := s.clusters[0].FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: ns, + TaskQueue: taskQueue, + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) + s.Require().NotEmpty(response.TaskToken) + return response +} + +func (s *xdcBaseSuite) completeActivityTask(ctx context.Context, task *workflowservice.PollActivityTaskQueueResponse) { + s.T().Helper() + _, err := s.clusters[0].FrontendClient().RespondActivityTaskCompleted(ctx, &workflowservice.RespondActivityTaskCompletedRequest{ + TaskToken: task.TaskToken, + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) +} + +func (s *xdcBaseSuite) failActivityTask(ctx context.Context, task *workflowservice.PollActivityTaskQueueResponse) { + s.T().Helper() + _, err := s.clusters[0].FrontendClient().RespondActivityTaskFailed(ctx, &workflowservice.RespondActivityTaskFailedRequest{ + TaskToken: task.TaskToken, + Identity: "buffered-events-xdc-test", + Failure: &failurepb.Failure{ + Message: "expected activity failure", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{NonRetryable: true}, + }, + }, + }) + s.Require().NoError(err) +} + +func (s *xdcBaseSuite) cancelActivityTask(ctx context.Context, task *workflowservice.PollActivityTaskQueueResponse) { + s.T().Helper() + _, err := s.clusters[0].FrontendClient().RespondActivityTaskCanceled(ctx, &workflowservice.RespondActivityTaskCanceledRequest{ + TaskToken: task.TaskToken, + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) +} + +func findActivityScheduledEventID(history []*historypb.HistoryEvent, activityID string) int64 { + for _, event := range history { + if event.EventType == enumspb.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED && + event.GetActivityTaskScheduledEventAttributes().GetActivityId() == activityID { + return event.EventId + } + } + return common.EmptyEventID +} + +func assertOnlyExpectedBufferedEventsReapplied(t require.TestingT, history []*historypb.HistoryEvent, bufferedTypes []enumspb.EventType) { + expectedCounts := map[enumspb.EventType]int{ + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED: 2, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ADMITTED: 1, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCEL_REQUESTED: 1, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED: 1, + } + for _, eventType := range bufferedTypes { + require.Equal(t, expectedCounts[eventType], countBufferedEventType(history, eventType), eventType.String()) + } +} + +func (s *FunctionalClustersTestSuite) startWorkflowWithPendingActivity( + ctx context.Context, + ns string, +) (*commonpb.WorkflowExecution, *taskqueuepb.TaskQueue) { + s.T().Helper() + workflowID := "buffered-events-xdc-" + uuid.NewString() + taskQueue := &taskqueuepb.TaskQueue{Name: "buffered-events-xdc", Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + startResponse, err := s.clusters[0].FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{ + Namespace: ns, + WorkflowId: workflowID, + WorkflowType: &commonpb.WorkflowType{Name: "buffered-events-xdc"}, + TaskQueue: taskQueue, + RequestId: uuid.NewString(), + WorkflowRunTimeout: durationpb.New(time.Minute), + WorkflowTaskTimeout: durationpb.New(30 * time.Second), + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) + execution := &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: startResponse.RunId} + + workflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + _, err = s.clusters[0].FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{ + TaskToken: workflowTask.TaskToken, + Identity: "buffered-events-xdc-test", + Commands: []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK, + Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{ + ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{ + ActivityId: "buffered-activity", + ActivityType: &commonpb.ActivityType{Name: "buffered-activity"}, + TaskQueue: taskQueue, + ScheduleToCloseTimeout: durationpb.New(time.Minute), + ScheduleToStartTimeout: durationpb.New(time.Minute), + StartToCloseTimeout: durationpb.New(time.Minute), + }, + }, + }}, + }) + s.Require().NoError(err) + return execution, taskQueue +} + +func (s *FunctionalClustersTestSuite) acceptUpdateAndStartTimer( + ctx context.Context, + ns string, + execution *commonpb.WorkflowExecution, + taskQueue *taskqueuepb.TaskQueue, + updateID string, +) { + s.T().Helper() + sdkClient, err := sdkclient.Dial(sdkclient.Options{ + HostPort: s.clusters[0].Host().FrontendGRPCAddress(), + Namespace: ns, + }) + s.Require().NoError(err) + defer sdkClient.Close() + updateCtx, cancelUpdate := context.WithCancel(ctx) + defer cancelUpdate() + updateResult := make(chan error, 1) + go func() { + _, updateErr := sdkClient.UpdateWorkflow(updateCtx, sdkclient.UpdateWorkflowOptions{ + UpdateID: updateID, + WorkflowID: execution.WorkflowId, + RunID: execution.RunId, + UpdateName: "buffered-update", + Args: []any{"source"}, + WaitForStage: sdkclient.WorkflowUpdateStageAccepted, + }) + updateResult <- updateErr + }() + await.Require(ctx, s.T(), func(t *await.T) { + response, pollErr := sdkClient.WorkflowService().PollWorkflowExecutionUpdate(t.Context(), &workflowservice.PollWorkflowExecutionUpdateRequest{ + Namespace: ns, + UpdateRef: &updatepb.UpdateRef{ + WorkflowExecution: execution, + UpdateId: updateID, + }, + WaitPolicy: &updatepb.WaitPolicy{ + LifecycleStage: enumspb.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ADMITTED, + }, + }) + require.NoError(t, pollErr) + require.Equal(t, enumspb.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ADMITTED, response.Stage) + }, 10*time.Second, 20*time.Millisecond) + workflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + s.Require().Len(workflowTask.Messages, 1) + updateRequestMessage := workflowTask.Messages[0] + s.Require().Equal(updateID, updateRequestMessage.ProtocolInstanceId) + acceptMessageID := "accept-" + uuid.NewString() + _, err = s.clusters[0].FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{ + TaskToken: workflowTask.TaskToken, + Identity: "buffered-events-xdc-test", + ForceCreateNewWorkflowTask: true, + Commands: []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_PROTOCOL_MESSAGE, + Attributes: &commandpb.Command_ProtocolMessageCommandAttributes{ + ProtocolMessageCommandAttributes: &commandpb.ProtocolMessageCommandAttributes{MessageId: acceptMessageID}, + }, + }, { + CommandType: enumspb.COMMAND_TYPE_START_TIMER, + Attributes: &commandpb.Command_StartTimerCommandAttributes{ + StartTimerCommandAttributes: &commandpb.StartTimerCommandAttributes{ + TimerId: "losing-branch-timer", + StartToFireTimeout: durationpb.New(100 * time.Millisecond), + }, + }, + }}, + Messages: []*protocolpb.Message{{ + Id: acceptMessageID, + ProtocolInstanceId: updateID, + Body: protoutils.MarshalAny(s.T(), &updatepb.Acceptance{ + AcceptedRequestMessageId: updateRequestMessage.Id, + AcceptedRequestSequencingEventId: updateRequestMessage.GetEventId(), + }), + }}, + }) + s.Require().NoError(err) + s.Require().True(hasUpdateAccepted(s.getWorkflowHistory(ctx, s.T(), 0, ns, execution), updateID)) + s.Require().NoError(<-updateResult) +} + +func (s *FunctionalClustersTestSuite) completeActivityAndBufferExternalEvents( + ctx context.Context, + ns string, + execution *commonpb.WorkflowExecution, + taskQueue *taskqueuepb.TaskQueue, +) string { + s.T().Helper() + activityTask, err := s.clusters[0].FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: ns, + TaskQueue: taskQueue, + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) + s.Require().NotEmpty(activityTask.TaskToken) + _, err = s.clusters[0].FrontendClient().RespondActivityTaskCompleted(ctx, &workflowservice.RespondActivityTaskCompletedRequest{ + TaskToken: activityTask.TaskToken, + Result: payloads.EncodeString("activity-result"), + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) + + _, err = s.clusters[0].FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: ns, + WorkflowExecution: execution, + SignalName: "buffered-signal", + Input: payloads.EncodeString("source"), + Identity: "buffered-events-xdc-test", + RequestId: uuid.NewString(), + }) + s.Require().NoError(err) + _, err = s.clusters[0].FrontendClient().RequestCancelWorkflowExecution(ctx, &workflowservice.RequestCancelWorkflowExecutionRequest{ + Namespace: ns, + WorkflowExecution: execution, + Identity: "buffered-events-xdc-test", + RequestId: uuid.NewString(), + }) + s.Require().NoError(err) + + optionsRequestID := uuid.NewString() + attachResponse, err := s.clusters[0].FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{ + Namespace: ns, + WorkflowId: execution.WorkflowId, + WorkflowType: &commonpb.WorkflowType{Name: "buffered-events-xdc"}, + TaskQueue: taskQueue, + RequestId: optionsRequestID, + WorkflowRunTimeout: durationpb.New(time.Minute), + WorkflowTaskTimeout: durationpb.New(30 * time.Second), + Identity: "buffered-events-xdc-test", + WorkflowIdConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING, + OnConflictOptions: &workflowpb.OnConflictOptions{ + AttachRequestId: true, + }, + }) + s.Require().NoError(err) + s.Require().False(attachResponse.Started) + + _, err = s.clusters[0].FrontendClient().PauseWorkflowExecution(ctx, &workflowservice.PauseWorkflowExecutionRequest{ + Namespace: ns, + WorkflowId: execution.WorkflowId, + RunId: execution.RunId, + Identity: "buffered-events-xdc-test", + Reason: "exercise buffered pause", + RequestId: uuid.NewString(), + }) + s.Require().NoError(err) + _, err = s.clusters[0].FrontendClient().UnpauseWorkflowExecution(ctx, &workflowservice.UnpauseWorkflowExecutionRequest{ + Namespace: ns, + WorkflowId: execution.WorkflowId, + RunId: execution.RunId, + Identity: "buffered-events-xdc-test", + Reason: "exercise buffered unpause", + RequestId: uuid.NewString(), + }) + s.Require().NoError(err) + return optionsRequestID +} + +func (s *xdcBaseSuite) failoverToNewActiveCluster(ctx context.Context, ns string) { + s.T().Helper() + response, err := s.clusters[0].FrontendClient().UpdateNamespace(ctx, &workflowservice.UpdateNamespaceRequest{ + Namespace: ns, + ReplicationConfig: &replicationpb.NamespaceReplicationConfig{ + ActiveClusterName: s.clusters[1].ClusterName(), + }, + }) + s.Require().NoError(err) + s.Require().Equal(int64(2), response.FailoverVersion) + await.Require(ctx, s.T(), func(t *await.T) { + for _, cluster := range s.clusters { + describeResponse, describeErr := cluster.FrontendClient().DescribeNamespace(t.Context(), &workflowservice.DescribeNamespaceRequest{ + Namespace: ns, + }) + require.NoError(t, describeErr) + require.Equal(t, s.clusters[1].ClusterName(), describeResponse.ReplicationConfig.ActiveClusterName) + } + }, replicationWaitTime, replicationCheckInterval) + s.waitForNamespaceCacheRefresh() +} + +func (s *xdcBaseSuite) writeSignalOnNewActive( + ctx context.Context, + signal activeClusterSignal, +) { + s.T().Helper() + request := &historyservice.SignalWorkflowExecutionRequest{ + NamespaceId: signal.Namespace.ID, + SignalRequest: &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: signal.Namespace.Name, + WorkflowExecution: signal.Execution, + SignalName: signal.SignalName, + Identity: "buffered-events-xdc-test", + RequestId: uuid.NewString(), + }, + } + await.Require(ctx, s.T(), func(t *await.T) { + _, err := s.clusters[1].HistoryClient().SignalWorkflowExecution(t.Context(), request) + require.NoError(t, err) + }, replicationWaitTime, replicationCheckInterval) + await.Require(ctx, s.T(), func(t *await.T) { + history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, signal.Namespace.Name, signal.Execution) + require.True(t, hasSignalNamed(history, signal.SignalName)) + }, replicationWaitTime, replicationCheckInterval) +} + +func (s *xdcBaseSuite) hasBufferedEventType( + ctx context.Context, + clusterIndex int, + ns string, + execution *commonpb.WorkflowExecution, + eventType enumspb.EventType, +) bool { + response, err := s.clusters[clusterIndex].AdminClient().DescribeMutableState(ctx, &adminservice.DescribeMutableStateRequest{ + Namespace: ns, + Execution: execution, + Archetype: chasm.WorkflowArchetype, + }) + if err != nil { + return false + } + return findHistoryEvent(response.GetCacheMutableState().GetBufferedEvents(), eventType, nil) != nil +} + +func (s *xdcBaseSuite) pollBufferedEventsWorkflowTask( + ctx context.Context, + clusterIndex int, + ns string, + taskQueue *taskqueuepb.TaskQueue, +) *workflowservice.PollWorkflowTaskQueueResponse { + s.T().Helper() + response, err := s.clusters[clusterIndex].FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{ + Namespace: ns, + TaskQueue: taskQueue, + Identity: "buffered-events-xdc-test", + }) + s.Require().NoError(err) + return response +} + +func (s *xdcBaseSuite) completeWorkflowTask(ctx context.Context, completion workflowTaskCompletion) { + s.T().Helper() + _, err := s.clusters[0].FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{ + TaskToken: completion.Task.TaskToken, + Identity: "buffered-events-xdc-test", + Commands: completion.Commands, + }) + s.Require().NoError(err) +} + +func (s *xdcBaseSuite) completeWorkflowTaskAndScheduleNext(ctx context.Context, completion workflowTaskCompletion) { + s.T().Helper() + _, err := s.clusters[0].FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{ + TaskToken: completion.Task.TaskToken, + Identity: "buffered-events-xdc-test", + Commands: completion.Commands, + ForceCreateNewWorkflowTask: true, + }) + s.Require().NoError(err) +} + +func (s *xdcBaseSuite) completeWorkflowTaskAndReturnNext( + ctx context.Context, + completion workflowTaskCompletion, +) *workflowservice.PollWorkflowTaskQueueResponse { + s.T().Helper() + response, err := s.clusters[0].FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{ + TaskToken: completion.Task.TaskToken, + Identity: "buffered-events-xdc-test", + Commands: completion.Commands, + ForceCreateNewWorkflowTask: true, + ReturnNewWorkflowTask: true, + }) + s.Require().NoError(err) + s.Require().NotNil(response.GetWorkflowTask()) + return response.GetWorkflowTask() +} + +func (s *xdcBaseSuite) blockReplicationForWorkflow( + clusterIndex int, + workflowID string, +) <-chan *blockedReplicationTask { + s.T().Helper() + // The interceptor runs on the receiving cluster, so clusterIndex identifies the replication + // destination rather than the cluster that generated the task. + tasks := make(chan *blockedReplicationTask, 20) + s.clusters[clusterIndex].InjectHook( + s.T(), + testhooks.NewHook(testhooks.HistoryReplicationTaskInterceptor, func( + task *replicationspb.ReplicationTask, + execute func() error, + ) error { + if workflowIDFromReplicationTask(task) != workflowID { + return execute() + } + blockedTask := &blockedReplicationTask{ + execute: execute, + result: make(chan error, 1), + } + tasks <- blockedTask + return <-blockedTask.result + }), + testhooks.GlobalScope, + ) + return tasks +} + +func (s *xdcBaseSuite) releaseReplicationTask( + ctx context.Context, + tasks <-chan *blockedReplicationTask, +) { + s.T().Helper() + select { + case task := <-tasks: + s.executeReplicationTask(task) + case <-ctx.Done(): + s.FailNow("timed out waiting for controlled history replication task", ctx.Err().Error()) + } +} + +func (s *xdcBaseSuite) tryReleaseReplicationTask(tasks <-chan *blockedReplicationTask) bool { + select { + case task := <-tasks: + s.executeReplicationTask(task) + return true + default: + return false + } +} + +func (s *xdcBaseSuite) executeReplicationTask(task *blockedReplicationTask) { + err := task.execute() + task.result <- err + var duplicateError *serviceerror.AlreadyExists + var retryReplicationError *serviceerrors.RetryReplication + s.Require().True( + err == nil || errors.As(err, &duplicateError) || errors.As(err, &retryReplicationError), + "replication task failed: %v", + err, + ) +} + +func workflowIDFromReplicationTask(task *replicationspb.ReplicationTask) string { + if attributes := task.GetSyncVersionedTransitionTaskAttributes(); attributes != nil { + return attributes.WorkflowId + } + if attributes := task.GetSyncWorkflowStateTaskAttributes(); attributes != nil { + return attributes.GetWorkflowState().GetExecutionInfo().GetWorkflowId() + } + if attributes := task.GetSyncHsmAttributes(); attributes != nil { + return attributes.WorkflowId + } + if attributes := task.GetSyncActivityTaskAttributes(); attributes != nil { + return attributes.WorkflowId + } + if attributes := task.GetVerifyVersionedTransitionTaskAttributes(); attributes != nil { + return attributes.WorkflowId + } + if attributes := task.GetBackfillHistoryTaskAttributes(); attributes != nil { + return attributes.WorkflowId + } + if attributes := task.GetHistoryTaskAttributes(); attributes != nil { + return attributes.WorkflowId + } + return "" +} + +func (s *xdcBaseSuite) assertBufferedEventTypes( + ctx context.Context, + expectation bufferedEventExpectation, +) { + s.T().Helper() + await.Require(ctx, s.T(), func(t *await.T) { + response, err := s.clusters[0].AdminClient().DescribeMutableState(t.Context(), &adminservice.DescribeMutableStateRequest{ + Namespace: expectation.Namespace, + Execution: expectation.Execution, + Archetype: chasm.WorkflowArchetype, + }) + require.NoError(t, err) + require.NotNil(t, response.CacheMutableState) + bufferedEvents := response.CacheMutableState.BufferedEvents + actualTypes := make([]enumspb.EventType, 0, len(bufferedEvents)) + for _, event := range bufferedEvents { + require.Equal(t, common.BufferedEventID, event.EventId) + actualTypes = append(actualTypes, event.EventType) + } + require.ElementsMatch(t, expectation.EventTypes, actualTypes) + }, replicationWaitTime, replicationCheckInterval) +} + +func (s *xdcBaseSuite) assertBufferedEventTypesPresent( + ctx context.Context, + expectation bufferedEventExpectation, +) { + s.T().Helper() + await.Require(ctx, s.T(), func(t *await.T) { + response, err := s.clusters[0].AdminClient().DescribeMutableState(t.Context(), &adminservice.DescribeMutableStateRequest{ + Namespace: expectation.Namespace, + Execution: expectation.Execution, + Archetype: chasm.WorkflowArchetype, + }) + require.NoError(t, err) + bufferedEvents := response.GetCacheMutableState().GetBufferedEvents() + for _, event := range bufferedEvents { + require.Equal(t, common.BufferedEventID, event.EventId) + } + for _, expectedType := range expectation.EventTypes { + require.NotNil(t, findHistoryEvent(bufferedEvents, expectedType, nil), "%s must be naturally buffered", expectedType) + } + }, replicationWaitTime, replicationCheckInterval) +} + +func (s *xdcBaseSuite) assertNoBufferedEvents( + ctx context.Context, + clusterIndex int, + ns string, + execution *commonpb.WorkflowExecution, +) { + s.T().Helper() + response, err := s.clusters[clusterIndex].AdminClient().DescribeMutableState(ctx, &adminservice.DescribeMutableStateRequest{ + Namespace: ns, + Execution: execution, + Archetype: chasm.WorkflowArchetype, + }) + s.Require().NoError(err) + s.Require().Empty(response.GetCacheMutableState().GetBufferedEvents()) + s.Require().Empty(response.GetDatabaseMutableState().GetBufferedEvents()) +} + +func (s *xdcBaseSuite) getWorkflowHistory( + ctx context.Context, + t require.TestingT, + clusterIndex int, + ns string, + execution *commonpb.WorkflowExecution, +) []*historypb.HistoryEvent { + response, err := s.clusters[clusterIndex].FrontendClient().GetWorkflowExecutionHistory(ctx, &workflowservice.GetWorkflowExecutionHistoryRequest{ + Namespace: ns, + Execution: execution, + }) + require.NoError(t, err) + return response.History.Events +} + +func (s *FunctionalClustersTestSuite) assertBufferedEventsPersistedOnLosingBranch( + ctx context.Context, + expectation bufferedInputsExpectation, +) { + s.T().Helper() + losingHistory := s.findNonCurrentHistoryBranch(ctx, expectation.Namespace.Name, expectation.Namespace.ID, expectation.Execution, func(history []*historypb.HistoryEvent) bool { + return hasSignalNamed(history, "buffered-signal") + }) + s.Require().True(hasUpdateAccepted(losingHistory, expectation.UpdateID)) + s.Require().True(hasWorkflowTaskFailedForFailover(losingHistory)) + s.Require().True(hasOptionsUpdatedRequest(losingHistory, expectation.OptionsRequestID)) + + for _, eventType := range expectation.EventTypes { + event := findHistoryEvent(losingHistory, eventType, nil) + s.Require().NotNil(event, "%s must be written to the losing branch", eventType) + s.Require().NotEqual(common.BufferedEventID, event.EventId) + s.Require().Positive(event.EventId) + } +} + +func (s *xdcBaseSuite) findNonCurrentHistoryBranch( + ctx context.Context, + ns string, + namespaceID string, + execution *commonpb.WorkflowExecution, + matches func([]*historypb.HistoryEvent) bool, +) []*historypb.HistoryEvent { + s.T().Helper() + description, err := s.clusters[0].AdminClient().DescribeMutableState(ctx, &adminservice.DescribeMutableStateRequest{ + Namespace: ns, + Execution: execution, + Archetype: chasm.WorkflowArchetype, + }) + s.Require().NoError(err) + versionHistories := description.GetDatabaseMutableState().GetExecutionInfo().GetVersionHistories() + s.Require().GreaterOrEqual(len(versionHistories.GetHistories()), 2) + + shardID := common.WorkflowIDToHistoryShard(namespaceID, execution.WorkflowId, s.numHistoryShards) + for index, versionHistory := range versionHistories.GetHistories() { + if int32(index) == versionHistories.GetCurrentVersionHistoryIndex() { + continue + } + history := s.readBufferedEventsHistoryBranch(ctx, shardID, versionHistory.GetBranchToken()) + if matches(history) { + return history + } + } + s.FailNow("matching non-current history branch not found") + return nil +} + +func (s *xdcBaseSuite) readBufferedEventsHistoryBranch( + ctx context.Context, + shardID int32, + branchToken []byte, +) []*historypb.HistoryEvent { + s.T().Helper() + request := &persistence.ReadHistoryBranchRequest{ + ShardID: shardID, + BranchToken: branchToken, + MinEventID: common.FirstEventID, + MaxEventID: common.EndEventID, + PageSize: 1000, + } + var events []*historypb.HistoryEvent + for { + response, err := s.clusters[0].ExecutionManager().ReadHistoryBranch(ctx, request) + s.Require().NoError(err) + events = append(events, response.HistoryEvents...) + if len(response.NextPageToken) == 0 { + return events + } + request.NextPageToken = response.NextPageToken + } +} + +func (s *xdcBaseSuite) bufferedEventsHistoriesEqual( + ctx context.Context, + ns string, + execution *commonpb.WorkflowExecution, +) bool { + sourceHistory := s.getWorkflowHistory(ctx, s.T(), 0, ns, execution) + targetHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution) + if len(targetHistory) != len(sourceHistory) { + return false + } + for index := range targetHistory { + if !proto.Equal(targetHistory[index], sourceHistory[index]) { + return false + } + } + return true +} + +func (s *FunctionalClustersTestSuite) hasReappliedBufferedInputs( + ctx context.Context, + expectation bufferedInputsExpectation, +) bool { + history := s.getWorkflowHistory(ctx, s.T(), 1, expectation.Namespace.Name, expectation.Execution) + hasSourceSignal := false + hasUpdate := false + hasCancel := false + hasOptionsUpdate := false + for _, event := range history { + switch event.EventType { + case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED: + hasSourceSignal = hasSourceSignal || event.GetWorkflowExecutionSignaledEventAttributes().SignalName == "buffered-signal" + case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ADMITTED: + hasUpdate = hasUpdate || event.GetWorkflowExecutionUpdateAdmittedEventAttributes().GetRequest().GetMeta().GetUpdateId() == expectation.UpdateID + case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCEL_REQUESTED: + hasCancel = true + case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED: + hasOptionsUpdate = hasOptionsUpdate || event.GetWorkflowExecutionOptionsUpdatedEventAttributes().AttachedRequestId == expectation.OptionsRequestID + default: + } + } + return hasSourceSignal && hasUpdate && hasCancel && hasOptionsUpdate +} + +func countBufferedEventType(history []*historypb.HistoryEvent, eventType enumspb.EventType) int { + count := 0 + for _, event := range history { + if event.EventType == eventType { + count++ + } + } + return count +} + +func hasSignalNamed(history []*historypb.HistoryEvent, signalName string) bool { + for _, event := range history { + if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED && + event.GetWorkflowExecutionSignaledEventAttributes().SignalName == signalName { + return true + } + } + return false +} + +func hasUpdateAccepted(history []*historypb.HistoryEvent, updateID string) bool { + for _, event := range history { + if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ACCEPTED && + event.GetWorkflowExecutionUpdateAcceptedEventAttributes().ProtocolInstanceId == updateID { + return true + } + } + return false +} + +func hasWorkflowTaskFailedForFailover(history []*historypb.HistoryEvent) bool { + for _, event := range history { + if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_TASK_FAILED && + event.GetWorkflowTaskFailedEventAttributes().Cause == enumspb.WORKFLOW_TASK_FAILED_CAUSE_FAILOVER_CLOSE_COMMAND { + return true + } + } + return false +} + +func hasOptionsUpdatedRequest(history []*historypb.HistoryEvent, requestID string) bool { + for _, event := range history { + if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED && + event.GetWorkflowExecutionOptionsUpdatedEventAttributes().AttachedRequestId == requestID { + return true + } + } + return false +} + +func (s *NexusStateReplicationSuite) createBufferedNexusEndpoint(ctx context.Context, handler nexustest.Handler) string { + s.T().Helper() + listenAddress := nexustest.AllocListenAddress() + nexustest.NewNexusServer(s.T(), listenAddress, handler) + for _, cluster := range s.clusters { + cluster.OverrideDynamicConfig( + s.T(), + nexusoperations.CallbackURLTemplate, + "http://"+s.clusters[0].Host().FrontendHTTPAddress()+"/namespaces/{{.NamespaceName}}/nexus/callback", + ) + } + endpointName := testcore.RandomizedNexusEndpoint(s.T().Name()) + for _, cluster := range s.clusters { + _, err := cluster.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{ + Spec: &nexuspb.EndpointSpec{ + Name: endpointName, + Target: &nexuspb.EndpointTarget{Variant: &nexuspb.EndpointTarget_External_{ + External: &nexuspb.EndpointTarget_External{Url: "http://" + listenAddress}, + }}, + }, + }) + s.Require().NoError(err) + } + return endpointName +} + +func (s *NexusStateReplicationSuite) failNexusOperation(ctx context.Context, callback bufferedNexusCallback) { + client := nexusrpc.NewCompletionHTTPClient(nexusrpc.CompletionHTTPClientOptions{Serializer: commonnexus.PayloadSerializer}) + err := client.CompleteOperation(ctx, callback.url, nexusrpc.CompleteOperationOptions{ + Error: &nexus.OperationError{ + State: nexus.OperationStateFailed, + Cause: &nexus.FailureError{Failure: nexus.Failure{Message: "expected operation failure"}}, + }, + Header: nexus.Header{commonnexus.CallbackTokenHeader: callback.token}, + }) + s.Require().NoError(err) +} + +func (s *NexusStateReplicationSuite) completeBufferedNexusOperation(ctx context.Context, completion bufferedNexusOperationCompletion) { + s.T().Helper() + s.completeNexusOperation(ctx, completion.Result, completion.Callback.url, completion.Callback.token) +} + +func (s *NexusStateReplicationSuite) cancelBufferedNexusOperation(ctx context.Context, callback bufferedNexusCallback) { + s.T().Helper() + s.cancelNexusOperation(ctx, callback.url, callback.token) +} + +func (s *NexusStateReplicationSuite) setupBufferedNexusEndpoint( + ctx context.Context, +) (string, <-chan bufferedNexusCallback, chan struct{}) { + s.T().Helper() + operationCallbacks := make(chan bufferedNexusCallback, 2) + allowLosingOnlyOperationStart := make(chan struct{}) + handler := nexustest.Handler{ + OnStartOperation: func( + _ context.Context, + _, operation string, + _ *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + operationCallbacks <- bufferedNexusCallback{ + operation: operation, + url: options.CallbackURL, + token: options.CallbackHeader.Get(commonnexus.CallbackTokenHeader), + } + if operation == "losing-only-operation" { + select { + case <-allowLosingOnlyOperationStart: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return &nexus.HandlerStartOperationResultAsync{OperationToken: operation}, nil + }, + } + listenAddress := nexustest.AllocListenAddress() + nexustest.NewNexusServer(s.T(), listenAddress, handler) + for _, cluster := range s.clusters { + cluster.OverrideDynamicConfig( + s.T(), + nexusoperations.CallbackURLTemplate, + "http://"+s.clusters[0].Host().FrontendHTTPAddress()+"/namespaces/{{.NamespaceName}}/nexus/callback", + ) + } + + endpointName := testcore.RandomizedNexusEndpoint(s.T().Name()) + for _, cluster := range s.clusters { + _, err := cluster.OperatorClient().CreateNexusEndpoint(ctx, &operatorservice.CreateNexusEndpointRequest{ + Spec: &nexuspb.EndpointSpec{ + Name: endpointName, + Target: &nexuspb.EndpointTarget{Variant: &nexuspb.EndpointTarget_External_{ + External: &nexuspb.EndpointTarget_External{Url: "http://" + listenAddress}, + }}, + }, + }) + s.Require().NoError(err) + } + return endpointName, operationCallbacks, allowLosingOnlyOperationStart +} + +func scheduleBufferedNexusOperationCommand(endpoint, operation string) *commandpb.Command { + return scheduleBufferedNexusOperationCommandWithTimeout(endpoint, operation, time.Minute) +} + +func scheduleBufferedNexusOperationCommandWithTimeout(endpoint, operation string, timeout time.Duration) *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION, + Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{ + ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{ + Endpoint: endpoint, + Service: "service", + Operation: operation, + ScheduleToCloseTimeout: durationpb.New(timeout), + }, + }, + } +} + +func requestCancelNexusOperationCommand(scheduledEventID int64) *commandpb.Command { + return &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_REQUEST_CANCEL_NEXUS_OPERATION, + Attributes: &commandpb.Command_RequestCancelNexusOperationCommandAttributes{ + RequestCancelNexusOperationCommandAttributes: &commandpb.RequestCancelNexusOperationCommandAttributes{ + ScheduledEventId: scheduledEventID, + }, + }, + } +} + +func receiveBufferedNexusCallback( + ctx context.Context, + t require.TestingT, + operationCallbacks <-chan bufferedNexusCallback, + expectedOperation string, +) bufferedNexusCallback { + select { + case operationCallback := <-operationCallbacks: + require.Equal(t, expectedOperation, operationCallback.operation) + return operationCallback + case <-ctx.Done(): + require.FailNow(t, "timed out waiting for Nexus operation callback", expectedOperation) + return bufferedNexusCallback{} + } +} + +func receiveAnyBufferedNexusCallback( + ctx context.Context, + t require.TestingT, + operationCallbacks <-chan bufferedNexusCallback, +) bufferedNexusCallback { + select { + case operationCallback := <-operationCallbacks: + return operationCallback + case <-ctx.Done(): + require.FailNow(t, "timed out waiting for Nexus operation callback") + return bufferedNexusCallback{} + } +} + +func (s *NexusStateReplicationSuite) findBufferedNexusLosingBranch( + ctx context.Context, + branch bufferedNexusLosingBranch, +) []*historypb.HistoryEvent { + s.T().Helper() + history := s.findNonCurrentHistoryBranch(ctx, branch.Namespace.Name, branch.Namespace.ID, branch.Execution, func(history []*historypb.HistoryEvent) bool { + return hasNexusEventForScheduledID(history, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, branch.ScheduledEventID) + }) + for _, event := range history { + if event.EventType == enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED || + event.EventType == enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED { + s.Require().NotEqual(common.BufferedEventID, event.EventId) + s.Require().Positive(event.EventId) + } + } + return history +} + +func findNexusScheduledEventID(history []*historypb.HistoryEvent, operation string) int64 { + for _, event := range history { + if event.EventType == enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED && + event.GetNexusOperationScheduledEventAttributes().Operation == operation { + return event.EventId + } + } + return common.EmptyEventID +} + +func hasNexusEventForScheduledID(history []*historypb.HistoryEvent, eventType enumspb.EventType, scheduledEventID int64) bool { + for _, event := range history { + if event.EventType != eventType { + continue + } + var eventScheduledID int64 + switch eventType { + case enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED: + eventScheduledID = event.EventId + case enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED: + eventScheduledID = event.GetNexusOperationStartedEventAttributes().GetScheduledEventId() + case enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED: + eventScheduledID = event.GetNexusOperationCompletedEventAttributes().GetScheduledEventId() + case enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED: + eventScheduledID = event.GetNexusOperationFailedEventAttributes().GetScheduledEventId() + case enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED: + eventScheduledID = event.GetNexusOperationCanceledEventAttributes().GetScheduledEventId() + case enumspb.EVENT_TYPE_NEXUS_OPERATION_TIMED_OUT: + eventScheduledID = event.GetNexusOperationTimedOutEventAttributes().GetScheduledEventId() + case enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_COMPLETED: + eventScheduledID = event.GetNexusOperationCancelRequestCompletedEventAttributes().GetScheduledEventId() + case enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_FAILED: + eventScheduledID = event.GetNexusOperationCancelRequestFailedEventAttributes().GetScheduledEventId() + default: + return false + } + if eventScheduledID == scheduledEventID { + return true + } + } + return false +} diff --git a/tests/xdc/buffered_events_replication_test.go b/tests/xdc/buffered_events_replication_test.go new file mode 100644 index 0000000000..491172d3a5 --- /dev/null +++ b/tests/xdc/buffered_events_replication_test.go @@ -0,0 +1,451 @@ +package xdc + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/server/common/testing/await" +) + +// Four values in HistoryBuilder's buffered-event set cannot be naturally +// buffered on current main. EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_REJECTED, +// EVENT_TYPE_WORKFLOW_PROPERTIES_MODIFIED_EXTERNALLY, and +// EVENT_TYPE_ACTIVITY_PROPERTIES_MODIFIED_EXTERNALLY have no production emitter. +// EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED has an emitter, but +// its production precondition rejects every workflow with a pending workflow +// task, so it cannot enter the buffer. The tests below cover every other value +// through its production API, transfer task, timer, callback, or reapplier. + +// TestNaturallyBufferedInputsFlushedAndReappliedAfterFailover buffers an activity result, timer, +// cancel request, signal, options update, pause, and unpause behind one workflow task; conflict +// reapplication then buffers an update-admitted event on the winner. It expects external inputs and +// the update to reach the winner while activity, timer, and pause state remain only on the losing branch. +func (s *FunctionalClustersTestSuite) TestNaturallyBufferedInputsFlushedAndReappliedAfterFailover() { + if !s.enableTransitionHistory { + s.T().Skip("buffered event state-based replication requires transition history") + } + + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + + namespace := s.createBufferedEventsNamespace(ctx) + ns := namespace.Name + s.enableWorkflowPauseForTest() + + // Phase 1: establish identical history with one pending activity. + execution, taskQueue := s.startWorkflowWithPendingActivity(ctx, ns) + workflowID := execution.WorkflowId + s.waitForClusterSynced() + await.Require(ctx, s.T(), func(t *await.T) { + history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution) + require.NotEmpty(t, history) + require.Equal(t, enumspb.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED, history[len(history)-1].EventType) + }, replicationWaitTime, replicationCheckInterval) + + replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID) + replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID) + + // Phase 2: hold a workflow task and create the losing branch's buffered events. + updateID := "buffered-update-" + uuid.NewString() + s.acceptUpdateAndStartTimer(ctx, ns, execution, taskQueue, updateID) + + heldWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + s.Require().NotEmpty(heldWorkflowTask.TaskToken) + + optionsRequestID := s.completeActivityAndBufferExternalEvents(ctx, ns, execution, taskQueue) + naturallyBufferedTypes := []enumspb.EventType{ + enumspb.EVENT_TYPE_ACTIVITY_TASK_STARTED, + enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED, + enumspb.EVENT_TYPE_TIMER_FIRED, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCEL_REQUESTED, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED, + } + bufferedInputs := bufferedInputsExpectation{ + Namespace: namespace, + Execution: execution, + UpdateID: updateID, + OptionsRequestID: optionsRequestID, + EventTypes: naturallyBufferedTypes, + } + s.assertBufferedEventTypes(ctx, bufferedEventExpectation{ + Namespace: ns, + Execution: execution, + EventTypes: naturallyBufferedTypes, + }) + + // Phase 3: fail over and create a winning branch on the new active cluster. + s.failoverToNewActiveCluster(ctx, ns) + s.writeSignalOnNewActive(ctx, activeClusterSignal{ + Namespace: namespace, + Execution: execution, + SignalName: "winner-signal", + }) + winnerWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 1, ns, taskQueue) + s.Require().NotEmpty(winnerWorkflowTask.TaskToken) + + // Phase 4: resolve the conflict and verify losing-branch storage versus reapplication. + s.releaseReplicationTask(ctx, replicationToOldActive) + s.assertNoBufferedEvents(ctx, 0, ns, execution) + s.assertBufferedEventsPersistedOnLosingBranch(ctx, bufferedInputs) + + // Reapplying the losing update while the winner's workflow task is running + // naturally creates UpdateAdmitted as a buffered event. + for attempt := 0; attempt < 10 && !s.hasBufferedEventType(ctx, 1, ns, execution, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ADMITTED); attempt++ { + s.releaseReplicationTask(ctx, replicationToNewActive) + } + s.Require().True(s.hasBufferedEventType(ctx, 1, ns, execution, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ADMITTED)) + await.Require(ctx, s.T(), func(t *await.T) { + require.False(t, s.hasBufferedEventType(t.Context(), 1, ns, execution, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ADMITTED)) + }, 45*time.Second, replicationCheckInterval) + // The timeout flush and conflict resolution can produce more than one + // state-based task. Release only this workflow's tasks until all reapplied + // inputs reach the active cluster. + for attempt := 0; attempt < 10 && !s.hasReappliedBufferedInputs(ctx, bufferedInputs); attempt++ { + s.releaseReplicationTask(ctx, replicationToNewActive) + } + s.Require().True(s.hasReappliedBufferedInputs(ctx, bufferedInputs)) + for attempt := 0; attempt < 10 && !s.bufferedEventsHistoriesEqual(ctx, ns, execution); attempt++ { + s.releaseReplicationTask(ctx, replicationToOldActive) + } + + await.Require(ctx, s.T(), func(t *await.T) { + sourceHistory := s.getWorkflowHistory(t.Context(), t.AssertionT(), 0, ns, execution) + targetHistory := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution) + require.Equal(t, targetHistory, sourceHistory) + }, replicationWaitTime, replicationCheckInterval) + + finalHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution) + s.Require().Equal(1, countBufferedEventType(finalHistory, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ADMITTED)) + s.Require().Equal(1, countBufferedEventType(finalHistory, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCEL_REQUESTED)) + s.Require().Equal(1, countBufferedEventType(finalHistory, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED)) + s.Require().Equal(2, countBufferedEventType(finalHistory, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED)) + assertOnlyExpectedBufferedEventsReapplied(s.T(), finalHistory, naturallyBufferedTypes) +} + +// TestNaturallyBufferedActivityOutcomesFlushedToLosingBranch buffers started, completed, failed, +// timed-out, and canceled activity outcomes. It expects all outcomes on the losing branch and none +// to be reapplied to the winner, apart from the winner independently producing its own timeout. +func (s *FunctionalClustersTestSuite) TestNaturallyBufferedActivityOutcomesFlushedToLosingBranch() { + if !s.enableTransitionHistory { + s.T().Skip("buffered event state-based replication requires transition history") + } + + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + namespace := s.createBufferedEventsNamespace(ctx) + ns := namespace.Name + + workflowID := "buffered-activities-xdc-" + uuid.NewString() + workflowQueue := &taskqueuepb.TaskQueue{Name: workflowID + "-workflow"} + execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{ + Namespace: ns, + WorkflowID: workflowID, + TaskQueue: workflowQueue, + }) + firstTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, workflowQueue) + + activityQueues := map[string]*taskqueuepb.TaskQueue{} + var scheduleCommands []*commandpb.Command + for _, activityID := range []string{"started", "completed", "failed", "canceled"} { + activityQueues[activityID] = &taskqueuepb.TaskQueue{Name: workflowID + "-" + activityID} + scheduleCommands = append(scheduleCommands, scheduleActivityCommand(activityID, activityQueues[activityID], time.Minute)) + } + s.completeWorkflowTask(ctx, workflowTaskCompletion{ + Task: firstTask, + Commands: scheduleCommands, + }) + + // Start the activity that will be canceled before establishing the common + // prefix, so its cancellation request can be issued by a workflow command. + canceledTask := s.pollBufferedActivityTask(ctx, ns, activityQueues["canceled"]) + s.signalWorkflow(ctx, workflowSignal{ + Namespace: ns, + Execution: execution, + SignalName: "prepare-activity-cancellation", + }) + cancelCommandTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, workflowQueue) + s.Require().NotNil(cancelCommandTask.History) + canceledScheduledID := findActivityScheduledEventID(cancelCommandTask.History.Events, "canceled") + s.Require().Positive(canceledScheduledID) + s.completeWorkflowTaskAndScheduleNext(ctx, workflowTaskCompletion{ + Task: cancelCommandTask, + Commands: []*commandpb.Command{ + requestCancelActivityCommand(canceledScheduledID), + scheduleActivityCommand("timed-out", &taskqueuepb.TaskQueue{Name: workflowID + "-unpolled"}, 5*time.Second), + }, + }) + + s.waitForClusterSynced() + await.Require(ctx, s.T(), func(t *await.T) { + history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution) + require.Positive(t, findActivityScheduledEventID(history, "canceled")) + }, replicationWaitTime, replicationCheckInterval) + + replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID) + replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID) + heldWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, workflowQueue) + s.Require().NotEmpty(heldWorkflowTask.TaskToken) + + startedTask := s.pollBufferedActivityTask(ctx, ns, activityQueues["started"]) + s.Require().NotEmpty(startedTask.TaskToken) + completedTask := s.pollBufferedActivityTask(ctx, ns, activityQueues["completed"]) + s.completeActivityTask(ctx, completedTask) + failedTask := s.pollBufferedActivityTask(ctx, ns, activityQueues["failed"]) + s.failActivityTask(ctx, failedTask) + s.cancelActivityTask(ctx, canceledTask) + + expectedTypes := []enumspb.EventType{ + enumspb.EVENT_TYPE_ACTIVITY_TASK_STARTED, + enumspb.EVENT_TYPE_ACTIVITY_TASK_STARTED, + enumspb.EVENT_TYPE_ACTIVITY_TASK_STARTED, + enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED, + enumspb.EVENT_TYPE_ACTIVITY_TASK_FAILED, + enumspb.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT, + enumspb.EVENT_TYPE_ACTIVITY_TASK_CANCELED, + } + s.assertBufferedEventTypes(ctx, bufferedEventExpectation{ + Namespace: ns, + Execution: execution, + EventTypes: expectedTypes, + }) + s.finishNaturallyBufferedConflict(ctx, naturallyBufferedConflict{ + Namespace: namespace, + Execution: execution, + ReplicationToOldActive: replicationToOldActive, + ReplicationToNewActive: replicationToNewActive, + ExpectedEventTypes: expectedTypes, + WinnerSignal: "activity-winner-signal", + }) + winningHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution) + for _, eventType := range []enumspb.EventType{ + enumspb.EVENT_TYPE_ACTIVITY_TASK_STARTED, + enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED, + enumspb.EVENT_TYPE_ACTIVITY_TASK_FAILED, + enumspb.EVENT_TYPE_ACTIVITY_TASK_CANCELED, + } { + s.Require().Zero(countBufferedEventType(winningHistory, eventType), "%s must remain only on the losing branch", eventType) + } + s.Require().Equal( + 1, + countBufferedEventType(winningHistory, enumspb.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT), + "the winner may produce its own timeout, but must not contain a duplicate reapplied timeout", + ) +} + +// TestNaturallyBufferedChildWorkflowOutcomesFlushedToLosingBranch buffers child start failure, +// started, completed, failed, canceled, timed-out, and terminated callbacks for children created only +// on the losing branch. It expects every callback to be persisted there and skipped on the winner. +func (s *FunctionalClustersTestSuite) TestNaturallyBufferedChildWorkflowOutcomesFlushedToLosingBranch() { + if !s.enableTransitionHistory { + s.T().Skip("buffered event state-based replication requires transition history") + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + namespace := s.createBufferedEventsNamespace(ctx) + ns := namespace.Name + + workflowID := "buffered-children-xdc-" + uuid.NewString() + parentQueue := &taskqueuepb.TaskQueue{Name: workflowID + "-parent"} + execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{ + Namespace: ns, + WorkflowID: workflowID, + TaskQueue: parentQueue, + }) + firstTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, parentQueue) + + childIDs := map[string]string{} + childQueues := map[string]*taskqueuepb.TaskQueue{} + for _, outcome := range []string{"completed", "failed", "canceled", "timed-out", "terminated", "duplicate"} { + childIDs[outcome] = workflowID + "-" + outcome + childQueues[outcome] = &taskqueuepb.TaskQueue{Name: childIDs[outcome] + "-queue"} + } + duplicateExecution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{ + Namespace: ns, + WorkflowID: childIDs["duplicate"], + TaskQueue: childQueues["duplicate"], + }) + s.Require().NotEmpty(duplicateExecution.RunId) + + s.waitForClusterSynced() + await.Require(ctx, s.T(), func(t *await.T) { + history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution) + require.NotEmpty(t, history) + }, replicationWaitTime, replicationCheckInterval) + + replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID) + replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID) + var commands []*commandpb.Command + for _, outcome := range []string{"completed", "failed", "canceled", "timed-out", "terminated", "duplicate"} { + runTimeout := time.Minute + if outcome == "timed-out" { + runTimeout = 5 * time.Second + } + commands = append(commands, startChildWorkflowCommand(childIDs[outcome], childQueues[outcome], runTimeout)) + } + s.completeWorkflowTaskAndScheduleNext(ctx, workflowTaskCompletion{ + Task: firstTask, + Commands: commands, + }) + heldWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, parentQueue) + s.Require().NotEmpty(heldWorkflowTask.TaskToken) + + completedTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, childQueues["completed"]) + s.completeWorkflowTask(ctx, workflowTaskCompletion{ + Task: completedTask, + Commands: []*commandpb.Command{completeWorkflowCommand()}, + }) + + failedTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, childQueues["failed"]) + s.completeWorkflowTask(ctx, workflowTaskCompletion{ + Task: failedTask, + Commands: []*commandpb.Command{failWorkflowCommand("expected child failure")}, + }) + + s.requestWorkflowCancellationEventually(ctx, workflowCancellation{ + Namespace: ns, + Execution: &commonpb.WorkflowExecution{WorkflowId: childIDs["canceled"]}, + }) + canceledTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, childQueues["canceled"]) + s.completeWorkflowTask(ctx, workflowTaskCompletion{ + Task: canceledTask, + Commands: []*commandpb.Command{cancelWorkflowCommand()}, + }) + + s.terminateWorkflowEventually(ctx, workflowTermination{ + Namespace: ns, + Execution: &commonpb.WorkflowExecution{WorkflowId: childIDs["terminated"]}, + Reason: "expected child termination", + }) + + expectedTypes := []enumspb.EventType{ + enumspb.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_FAILED, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_FAILED, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_CANCELED, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_TIMED_OUT, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_TERMINATED, + } + s.assertBufferedEventTypesPresent(ctx, bufferedEventExpectation{ + Namespace: ns, + Execution: execution, + EventTypes: expectedTypes, + }) + s.finishNaturallyBufferedConflict(ctx, naturallyBufferedConflict{ + Namespace: namespace, + Execution: execution, + ReplicationToOldActive: replicationToOldActive, + ReplicationToNewActive: replicationToNewActive, + ExpectedEventTypes: expectedTypes, + WinnerSignal: "child-winner-signal", + }) + winningHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution) + for _, eventType := range expectedTypes { + s.Require().Zero(countBufferedEventType(winningHistory, eventType), "%s has no child initiation on the winning branch and must be skipped", eventType) + } +} + +// TestNaturallyBufferedExternalWorkflowOutcomesFlushedToLosingBranch buffers successful and failed +// signal-external and cancel-external results. It expects every result on the losing branch and none +// on the winner because the corresponding initiated commands do not exist there. +func (s *FunctionalClustersTestSuite) TestNaturallyBufferedExternalWorkflowOutcomesFlushedToLosingBranch() { + if !s.enableTransitionHistory { + s.T().Skip("buffered event state-based replication requires transition history") + } + + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + namespace := s.createBufferedEventsNamespace(ctx) + ns := namespace.Name + + workflowID := "buffered-external-xdc-" + uuid.NewString() + workflowQueue := &taskqueuepb.TaskQueue{Name: workflowID + "-source"} + execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{ + Namespace: ns, + WorkflowID: workflowID, + TaskQueue: workflowQueue, + }) + firstTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, workflowQueue) + signalTargetID := workflowID + "-signal-target" + signalTargetQueue := &taskqueuepb.TaskQueue{Name: signalTargetID + "-queue"} + signalTargetExecution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{ + Namespace: ns, + WorkflowID: signalTargetID, + TaskQueue: signalTargetQueue, + }) + s.Require().NotEmpty(signalTargetExecution.RunId) + targetTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, signalTargetQueue) + s.completeWorkflowTask(ctx, workflowTaskCompletion{ + Task: targetTask, + }) + cancelTargetID := workflowID + "-completed-cancel-target" + cancelTargetExecution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{ + Namespace: ns, + WorkflowID: cancelTargetID, + TaskQueue: &taskqueuepb.TaskQueue{Name: cancelTargetID + "-queue"}, + }) + s.terminateWorkflow(ctx, workflowTermination{ + Namespace: ns, + Execution: cancelTargetExecution, + Reason: "completed target makes external cancellation deterministic", + }) + + s.waitForClusterSynced() + await.Require(ctx, s.T(), func(t *await.T) { + history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution) + require.NotEmpty(t, history) + }, replicationWaitTime, replicationCheckInterval) + replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID) + replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID) + + missingWorkflowID := workflowID + "-missing" + missingRunID := uuid.NewString() + heldWorkflowTask := s.completeWorkflowTaskAndReturnNext(ctx, workflowTaskCompletion{ + Task: firstTask, + Commands: []*commandpb.Command{ + signalExternalWorkflowCommand(signalTargetID, "", "successful-external-signal"), + signalExternalWorkflowCommand(missingWorkflowID, missingRunID, "failed-external-signal"), + cancelExternalWorkflowCommand(missingWorkflowID, missingRunID), + cancelExternalWorkflowCommand(cancelTargetID, cancelTargetExecution.RunId), + }, + }) + s.Require().NotEmpty(heldWorkflowTask.TaskToken) + await.Require(ctx, s.T(), func(t *await.T) { + history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 0, ns, signalTargetExecution) + require.True(t, hasSignalNamed(history, "successful-external-signal")) + }, replicationWaitTime, replicationCheckInterval) + + expectedTypes := []enumspb.EventType{ + enumspb.EVENT_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED, + enumspb.EVENT_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED, + enumspb.EVENT_TYPE_EXTERNAL_WORKFLOW_EXECUTION_SIGNALED, + enumspb.EVENT_TYPE_EXTERNAL_WORKFLOW_EXECUTION_CANCEL_REQUESTED, + } + s.assertBufferedEventTypesPresent(ctx, bufferedEventExpectation{ + Namespace: ns, + Execution: execution, + EventTypes: expectedTypes, + }) + s.finishNaturallyBufferedConflict(ctx, naturallyBufferedConflict{ + Namespace: namespace, + Execution: execution, + ReplicationToOldActive: replicationToOldActive, + ReplicationToNewActive: replicationToNewActive, + ExpectedEventTypes: expectedTypes, + WinnerSignal: "external-winner-signal", + }) + winningHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution) + for _, eventType := range expectedTypes { + s.Require().Zero(countBufferedEventType(winningHistory, eventType), "%s must remain only on the losing branch", eventType) + } +} diff --git a/tests/xdc/buffered_nexus_events_replication_test.go b/tests/xdc/buffered_nexus_events_replication_test.go new file mode 100644 index 0000000000..28477ba8f7 --- /dev/null +++ b/tests/xdc/buffered_nexus_events_replication_test.go @@ -0,0 +1,384 @@ +package xdc + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/nexus-rpc/sdk-go/nexus" + "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" + enumspb "go.temporal.io/api/enums/v1" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" + commonnexus "go.temporal.io/server/common/nexus" + "go.temporal.io/server/common/nexus/nexustest" + "go.temporal.io/server/common/testing/await" +) + +// TestBufferedNexusEventsReapplySharedOperationAndSkipLosingOnlyOperation buffers completion of an +// operation shared by both branches plus start and completion of an operation created only on the +// losing branch. It expects all three events on the loser, only the shared completion on the winner, +// and the losing-only operation skipped. This covers temporalio/temporal#10986. +func (s *NexusStateReplicationSuite) TestBufferedNexusEventsReapplySharedOperationAndSkipLosingOnlyOperation() { + if !s.enableTransitionHistory || s.chasmEnabled { + s.T().Skip("this conflict-reapplication regression is specific to transition-history HSM Nexus operations") + } + + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + namespace := s.createBufferedEventsNamespace(ctx) + ns := namespace.Name + + // Phase 1: establish identical history with one shared, started Nexus operation. + endpointName, operationCallbacks, allowLosingOnlyOperationStart := s.setupBufferedNexusEndpoint(ctx) + + workflowID := "buffered-nexus-conflict-" + uuid.NewString() + taskQueue := &taskqueuepb.TaskQueue{Name: "buffered-nexus-conflict", Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{ + Namespace: ns, + WorkflowID: workflowID, + TaskQueue: taskQueue, + }) + + firstWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + s.completeWorkflowTask(ctx, workflowTaskCompletion{ + Task: firstWorkflowTask, + Commands: []*commandpb.Command{scheduleBufferedNexusOperationCommand(endpointName, "shared-operation")}, + }) + sharedOperationCallback := receiveBufferedNexusCallback(ctx, s.T(), operationCallbacks, "shared-operation") + await.Require(ctx, s.T(), func(t *await.T) { + history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 0, ns, execution) + sharedOperationScheduledEventID := findNexusScheduledEventID(history, "shared-operation") + require.Positive(t, sharedOperationScheduledEventID) + require.True(t, hasNexusEventForScheduledID(history, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, sharedOperationScheduledEventID)) + }, replicationWaitTime, replicationCheckInterval) + sharedOperationScheduledEventID := findNexusScheduledEventID(s.getWorkflowHistory(ctx, s.T(), 0, ns, execution), "shared-operation") + s.Require().Positive(sharedOperationScheduledEventID) + s.waitForClusterSynced() + await.Require(ctx, s.T(), func(t *await.T) { + history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution) + require.True(t, hasNexusEventForScheduledID(history, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, sharedOperationScheduledEventID)) + }, replicationWaitTime, replicationCheckInterval) + + replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID) + replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID) + + // Phase 2: hold a workflow task and buffer completions for shared and losing-only operations. + s.signalWorkflow(ctx, workflowSignal{ + Namespace: ns, + Execution: execution, + SignalName: "create-losing-operation", + }) + scheduleLosingOnlyOperationTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + s.completeWorkflowTaskAndScheduleNext(ctx, workflowTaskCompletion{ + Task: scheduleLosingOnlyOperationTask, + Commands: []*commandpb.Command{scheduleBufferedNexusOperationCommand(endpointName, "losing-only-operation")}, + }) + heldWorkflowTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + s.Require().NotEmpty(heldWorkflowTask.TaskToken) + losingOnlyOperationScheduledEventID := findNexusScheduledEventID(heldWorkflowTask.History.Events, "losing-only-operation") + s.Require().Positive(losingOnlyOperationScheduledEventID) + close(allowLosingOnlyOperationStart) + losingOnlyOperationCallback := receiveBufferedNexusCallback(ctx, s.T(), operationCallbacks, "losing-only-operation") + + s.completeBufferedNexusOperation(ctx, bufferedNexusOperationCompletion{ + Callback: sharedOperationCallback, + Result: "shared-result", + }) + s.completeBufferedNexusOperation(ctx, bufferedNexusOperationCompletion{ + Callback: losingOnlyOperationCallback, + Result: "losing-result", + }) + s.assertBufferedEventTypes(ctx, bufferedEventExpectation{ + Namespace: ns, + Execution: execution, + EventTypes: []enumspb.EventType{ + enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, + enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, + enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, + }, + }) + + // Phase 3: fail over and create a winning branch on the new active cluster. + s.failoverToNewActiveCluster(ctx, ns) + s.writeSignalOnNewActive(ctx, activeClusterSignal{ + Namespace: namespace, + Execution: execution, + SignalName: "nexus-winner-signal", + }) + + // Phase 4: resolve the conflict and verify losing-branch storage versus reapplication. + s.releaseReplicationTask(ctx, replicationToOldActive) + s.assertNoBufferedEvents(ctx, 0, ns, execution) + losingHistory := s.findBufferedNexusLosingBranch(ctx, bufferedNexusLosingBranch{ + Namespace: namespace, + Execution: execution, + ScheduledEventID: losingOnlyOperationScheduledEventID, + }) + s.Require().True(hasNexusEventForScheduledID(losingHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED, losingOnlyOperationScheduledEventID)) + s.Require().True(hasNexusEventForScheduledID(losingHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, losingOnlyOperationScheduledEventID)) + s.Require().Equal(2, countBufferedEventType(losingHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED)) + + for range 10 { + targetHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution) + if hasNexusEventForScheduledID(targetHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, sharedOperationScheduledEventID) { + break + } + s.releaseReplicationTask(ctx, replicationToNewActive) + } + targetHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution) + s.Require().True(hasNexusEventForScheduledID(targetHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, sharedOperationScheduledEventID)) + s.Require().False(hasNexusEventForScheduledID(targetHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED, losingOnlyOperationScheduledEventID)) + s.Require().False(hasNexusEventForScheduledID(targetHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED, losingOnlyOperationScheduledEventID)) + await.Require(ctx, s.T(), func(t *await.T) { + describeResponse, describeErr := s.clusters[1].FrontendClient().DescribeWorkflowExecution(t.Context(), &workflowservice.DescribeWorkflowExecutionRequest{ + Namespace: ns, + Execution: execution, + }) + require.NoError(t, describeErr) + require.Empty(t, describeResponse.PendingNexusOperations) + }, replicationWaitTime, replicationCheckInterval) + + for attempt := 0; attempt < 10 && !s.bufferedEventsHistoriesEqual(ctx, ns, execution); attempt++ { + s.releaseReplicationTask(ctx, replicationToOldActive) + } + await.Require(ctx, s.T(), func(t *await.T) { + require.True(t, s.bufferedEventsHistoriesEqual(t.Context(), ns, execution)) + }, replicationWaitTime, replicationCheckInterval) +} + +// TestNaturallyBufferedNexusOutcomesFlushedAndReapplied buffers failed, canceled, timed-out, and +// cancel-request-failed outcomes for operations shared by both branches. It expects the terminal operation +// outcomes to be reapplied to the winner and the non-cherry-pickable cancellation result to remain on the loser. +func (s *NexusStateReplicationSuite) TestNaturallyBufferedNexusOutcomesFlushedAndReapplied() { + if !s.enableTransitionHistory || s.chasmEnabled { + s.T().Skip("this conflict-reapplication regression is specific to transition-history HSM Nexus operations") + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + namespace := s.createBufferedEventsNamespace(ctx) + ns := namespace.Name + + callbacks := make(chan bufferedNexusCallback, 4) + allowCancellationResponse := make(chan struct{}) + handler := nexustest.Handler{ + OnStartOperation: func(_ context.Context, _, operation string, _ *nexus.LazyValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) { + callbacks <- bufferedNexusCallback{ + operation: operation, + url: options.CallbackURL, + token: options.CallbackHeader.Get(commonnexus.CallbackTokenHeader), + } + return &nexus.HandlerStartOperationResultAsync{OperationToken: operation}, nil + }, + OnCancelOperation: func(_ context.Context, _, operation, _ string, _ nexus.CancelOperationOptions) error { + <-allowCancellationResponse + if operation == "cancel-failed" { + return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "expected cancellation failure") + } + return nil + }, + } + endpointName := s.createBufferedNexusEndpoint(ctx, handler) + + workflowID := "buffered-nexus-outcomes-" + uuid.NewString() + taskQueue := &taskqueuepb.TaskQueue{Name: workflowID, Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{ + Namespace: ns, + WorkflowID: workflowID, + TaskQueue: taskQueue, + }) + firstTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + operations := []string{"failed", "canceled", "timed-out", "cancel-failed"} + commands := make([]*commandpb.Command, 0, len(operations)) + for _, operation := range operations { + timeout := time.Minute + if operation == "timed-out" { + timeout = 5 * time.Second + } + commands = append(commands, scheduleBufferedNexusOperationCommandWithTimeout(endpointName, operation, timeout)) + } + s.completeWorkflowTask(ctx, workflowTaskCompletion{ + Task: firstTask, + Commands: commands, + }) + + operationCallbacks := make(map[string]bufferedNexusCallback, len(operations)) + for range operations { + callback := receiveAnyBufferedNexusCallback(ctx, s.T(), callbacks) + operationCallbacks[callback.operation] = callback + } + await.Require(ctx, s.T(), func(t *await.T) { + history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 0, ns, execution) + for _, operation := range operations { + scheduledID := findNexusScheduledEventID(history, operation) + require.Positive(t, scheduledID) + require.True(t, hasNexusEventForScheduledID(history, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, scheduledID)) + } + }, replicationWaitTime, replicationCheckInterval) + s.waitForClusterSynced() + await.Require(ctx, s.T(), func(t *await.T) { + history := s.getWorkflowHistory(t.Context(), t.AssertionT(), 1, ns, execution) + for _, operation := range operations { + scheduledID := findNexusScheduledEventID(history, operation) + require.Positive(t, scheduledID) + require.True(t, hasNexusEventForScheduledID(history, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED, scheduledID)) + } + }, replicationWaitTime, replicationCheckInterval) + await.Require(ctx, s.T(), func(t *await.T) { + response, describeErr := s.clusters[1].FrontendClient().DescribeWorkflowExecution(t.Context(), &workflowservice.DescribeWorkflowExecutionRequest{ + Namespace: ns, + Execution: execution, + }) + require.NoError(t, describeErr) + require.Len(t, response.PendingNexusOperations, len(operations)) + }, replicationWaitTime, replicationCheckInterval) + replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID) + replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID) + + triggerTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + history := triggerTask.History.Events + scheduledIDs := make(map[string]int64, len(operations)) + for _, operation := range operations { + scheduledIDs[operation] = findNexusScheduledEventID(history, operation) + s.Require().Positive(scheduledIDs[operation]) + } + s.completeWorkflowTaskAndScheduleNext(ctx, workflowTaskCompletion{ + Task: triggerTask, + Commands: []*commandpb.Command{ + requestCancelNexusOperationCommand(findNexusScheduledEventID(history, "cancel-failed")), + }, + }) + heldTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + s.Require().NotEmpty(heldTask.TaskToken) + close(allowCancellationResponse) + + s.failNexusOperation(ctx, operationCallbacks["failed"]) + s.cancelBufferedNexusOperation(ctx, operationCallbacks["canceled"]) + expectedTypes := []enumspb.EventType{ + enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED, + enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED, + enumspb.EVENT_TYPE_NEXUS_OPERATION_TIMED_OUT, + enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_FAILED, + } + s.assertBufferedEventTypesPresent(ctx, bufferedEventExpectation{ + Namespace: ns, + Execution: execution, + EventTypes: expectedTypes, + }) + + losingHistory := s.finishNaturallyBufferedConflict(ctx, naturallyBufferedConflict{ + Namespace: namespace, + Execution: execution, + ReplicationToOldActive: replicationToOldActive, + ReplicationToNewActive: replicationToNewActive, + ExpectedEventTypes: expectedTypes, + WinnerSignal: "nexus-outcomes-winner-signal", + }) + for _, eventType := range expectedTypes { + s.Require().NotNil(findHistoryEvent(losingHistory, eventType, nil)) + } + winningHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution) + for operation, eventType := range map[string]enumspb.EventType{ + "failed": enumspb.EVENT_TYPE_NEXUS_OPERATION_FAILED, + "canceled": enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCELED, + "timed-out": enumspb.EVENT_TYPE_NEXUS_OPERATION_TIMED_OUT, + } { + s.Require().True( + hasNexusEventForScheduledID(winningHistory, eventType, scheduledIDs[operation]), + "%s for common operation %q must be reapplied to the winning branch", + eventType, + operation, + ) + } + s.Require().False( + hasNexusEventForScheduledID(winningHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_FAILED, scheduledIDs["cancel-failed"]), + "Nexus cancellation results are not cherry-pickable and must remain only on the losing branch", + ) +} + +// TestNaturallyBufferedNexusCancelRequestCompletedFlushedAndReapplied buffers a successful Nexus +// cancel-request result. It expects the result to be persisted on the losing branch but skipped on the +// winner because cancellation results are not cherry-pickable. +func (s *NexusStateReplicationSuite) TestNaturallyBufferedNexusCancelRequestCompletedFlushedAndReapplied() { + if !s.enableTransitionHistory || s.chasmEnabled { + s.T().Skip("this conflict-reapplication regression is specific to transition-history HSM Nexus operations") + } + + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + namespace := s.createBufferedEventsNamespace(ctx) + ns := namespace.Name + + started := make(chan struct{}, 1) + canceled := make(chan struct{}, 1) + allowCancellationResponse := make(chan struct{}) + handler := nexustest.Handler{ + OnStartOperation: func(_ context.Context, _, operation string, _ *nexus.LazyValue, _ nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) { + started <- struct{}{} + return &nexus.HandlerStartOperationResultAsync{OperationToken: operation}, nil + }, + OnCancelOperation: func(_ context.Context, _, _, _ string, _ nexus.CancelOperationOptions) error { + canceled <- struct{}{} + <-allowCancellationResponse + return nil + }, + } + endpointName := s.createBufferedNexusEndpoint(ctx, handler) + workflowID := "buffered-nexus-cancel-completed-" + uuid.NewString() + taskQueue := &taskqueuepb.TaskQueue{Name: workflowID, Kind: enumspb.TASK_QUEUE_KIND_NORMAL} + execution := s.startBufferedEventsWorkflow(ctx, startBufferedEventsWorkflowArgs{ + Namespace: ns, + WorkflowID: workflowID, + TaskQueue: taskQueue, + }) + firstTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + s.completeWorkflowTask(ctx, workflowTaskCompletion{ + Task: firstTask, + Commands: []*commandpb.Command{scheduleBufferedNexusOperationCommand(endpointName, "cancel-completed")}, + }) + select { + case <-started: + case <-ctx.Done(): + s.FailNow("timed out waiting for Nexus operation to start") + } + triggerTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + scheduledID := findNexusScheduledEventID(triggerTask.History.Events, "cancel-completed") + s.Require().Positive(scheduledID) + s.waitForClusterSynced() + replicationToOldActive := s.blockReplicationForWorkflow(0, workflowID) + replicationToNewActive := s.blockReplicationForWorkflow(1, workflowID) + s.completeWorkflowTaskAndScheduleNext(ctx, workflowTaskCompletion{ + Task: triggerTask, + Commands: []*commandpb.Command{requestCancelNexusOperationCommand(scheduledID)}, + }) + heldTask := s.pollBufferedEventsWorkflowTask(ctx, 0, ns, taskQueue) + s.Require().NotEmpty(heldTask.TaskToken) + close(allowCancellationResponse) + select { + case <-canceled: + case <-ctx.Done(): + s.FailNow("timed out waiting for Nexus cancellation handler") + } + expectedTypes := []enumspb.EventType{enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_COMPLETED} + s.assertBufferedEventTypesPresent(ctx, bufferedEventExpectation{ + Namespace: ns, + Execution: execution, + EventTypes: expectedTypes, + }) + + s.finishNaturallyBufferedConflict(ctx, naturallyBufferedConflict{ + Namespace: namespace, + Execution: execution, + ReplicationToOldActive: replicationToOldActive, + ReplicationToNewActive: replicationToNewActive, + ExpectedEventTypes: expectedTypes, + WinnerSignal: "nexus-cancel-completed-winner-signal", + }) + winningHistory := s.getWorkflowHistory(ctx, s.T(), 1, ns, execution) + s.Require().False( + hasNexusEventForScheduledID(winningHistory, enumspb.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_COMPLETED, scheduledID), + "Nexus cancellation results are not cherry-pickable and must remain only on the losing branch", + ) +}