diff --git a/tests/xdc/base.go b/tests/xdc/base.go index c3c6c6bbaf..ee5d891e37 100644 --- a/tests/xdc/base.go +++ b/tests/xdc/base.go @@ -25,6 +25,7 @@ import ( "go.temporal.io/server/common/log" "go.temporal.io/server/common/log/tag" "go.temporal.io/server/common/searchattribute" + "go.temporal.io/server/common/testing/await" "go.temporal.io/server/common/testing/historyrequire" "go.temporal.io/server/common/testing/protorequire" "go.temporal.io/server/tests/testcore" @@ -61,6 +62,7 @@ type ( startTime time.Time onceClusterConnect sync.Once + numHistoryShards int32 enableTransitionHistory bool @@ -82,6 +84,7 @@ func (s *xdcBaseSuite) clusterReplicationConfig() []*replicationpb.ClusterReplic func (s *xdcBaseSuite) setupSuite(opts ...testcore.TestClusterOption) { params := testcore.ApplyTestClusterOptions(opts) + s.numHistoryShards = cmp.Or(params.NumHistoryShards, int32(1)) if s.logger == nil { s.logger = log.NewTestLogger() @@ -139,7 +142,7 @@ func (s *xdcBaseSuite) setupSuite(opts ...testcore.TestClusterOption) { }, }, HistoryConfig: testcore.HistoryConfig{ - NumHistoryShards: cmp.Or(params.NumHistoryShards, 1), + NumHistoryShards: s.numHistoryShards, }, Persistence: persistenceDefaults, DynamicConfigOverrides: s.dynamicConfigOverrides, @@ -184,27 +187,29 @@ func (s *xdcBaseSuite) waitForClusterConnected( sourceCluster *testcore.TestCluster, targetClusterName string, ) { + expectedNumHistoryShards := cmp.Or(s.numHistoryShards, int32(1)) s.logger.Info("wait for clusters to be synced", tag.SourceCluster(sourceCluster.ClusterName()), tag.TargetCluster(targetClusterName)) - s.EventuallyWithT(func(c *assert.CollectT) { + await.Require(context.Background(), s.T(), func(c *await.T) { s.logger.Info("check if clusters are synced", tag.SourceCluster(sourceCluster.ClusterName()), tag.TargetCluster(targetClusterName)) - resp, err := sourceCluster.HistoryClient().GetReplicationStatus(context.Background(), &historyservice.GetReplicationStatusRequest{}) + resp, err := sourceCluster.HistoryClient().GetReplicationStatus(c.Context(), &historyservice.GetReplicationStatusRequest{}) require.NoError(c, err) - require.Lenf(c, resp.Shards, 1, "test cluster has only one history shard") + require.Lenf(c, resp.Shards, int(expectedNumHistoryShards), "unexpected history shard count") - shard := resp.Shards[0] - require.NotNil(c, shard) - require.Positive(c, shard.MaxReplicationTaskId) - require.NotNil(c, shard.ShardLocalTime) - require.WithinRange(c, shard.ShardLocalTime.AsTime(), s.startTime, time.Now()) - require.NotNil(c, shard.RemoteClusters) + for _, shard := range resp.Shards { + require.NotNil(c, shard) + require.Positive(c, shard.MaxReplicationTaskId) + require.NotNil(c, shard.ShardLocalTime) + require.WithinRange(c, shard.ShardLocalTime.AsTime(), s.startTime, time.Now()) + require.NotNil(c, shard.RemoteClusters) - standbyAckInfo, ok := shard.RemoteClusters[targetClusterName] - require.True(c, ok) - require.NotNil(c, standbyAckInfo) - require.LessOrEqual(c, shard.MaxReplicationTaskId, standbyAckInfo.AckedTaskId) - require.NotNil(c, standbyAckInfo.AckedTaskVisibilityTime) - require.WithinRange(c, standbyAckInfo.AckedTaskVisibilityTime.AsTime(), s.startTime, time.Now()) - }, 90*time.Second, 1*time.Second) + standbyAckInfo, ok := shard.RemoteClusters[targetClusterName] + require.True(c, ok) + require.NotNil(c, standbyAckInfo) + require.LessOrEqual(c, shard.MaxReplicationTaskId, standbyAckInfo.AckedTaskId) + require.NotNil(c, standbyAckInfo.AckedTaskVisibilityTime) + require.WithinRange(c, standbyAckInfo.AckedTaskVisibilityTime.AsTime(), s.startTime, time.Now()) + } + }, 90*time.Second, time.Second) s.logger.Info("clusters synced", tag.SourceCluster(sourceCluster.ClusterName()), tag.TargetCluster(targetClusterName)) } diff --git a/tests/xdc/parent_child_harness.go b/tests/xdc/parent_child_harness.go new file mode 100644 index 0000000000..4588e521e5 --- /dev/null +++ b/tests/xdc/parent_child_harness.go @@ -0,0 +1,1262 @@ +package xdc + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "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" + historypb "go.temporal.io/api/history/v1" + replicationpb "go.temporal.io/api/replication/v1" + "go.temporal.io/api/serviceerror" + "go.temporal.io/api/workflowservice/v1" + enumsspb "go.temporal.io/server/api/enums/v1" + "go.temporal.io/server/api/historyservice/v1" + persistencespb "go.temporal.io/server/api/persistence/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" + "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/metrics/metricstest" + "go.temporal.io/server/common/persistence/serialization" + "go.temporal.io/server/common/testing/await" + "go.temporal.io/server/common/testing/taskpoller" + "go.temporal.io/server/common/testing/testhooks" + "go.temporal.io/server/common/testing/testvars" + "go.temporal.io/server/tests/testcore" + "google.golang.org/protobuf/types/known/durationpb" +) + +var ( + errParentChildReplicationGateClosed = errors.New("parent-child replication gate is closed") + errParentChildTaskAlreadyResolved = errors.New("parent-child replication task is already resolved") +) + +type parentChildXDCTestSuite struct{ xdcBaseSuite } + +type ( + parentChildWorkflow int + + parentChildCluster int + + parentChildReplicationTaskAction int + + parentChildScenario struct { + steps []parentChildScenarioStep + expectations []parentChildExpectation + } + + parentChildScenarioStep struct { + name string + run func(context.Context, *parentChildScenarioRuntime) error + } + + parentChildExpectation struct { + name string + check func(context.Context, *parentChildScenarioRuntime) error + } + + parentChildScenarioRuntime struct { + suite *parentChildXDCTestSuite + + namespace string + namespaceID string + parentID string + parentRunID string + childID string + childRunID string + parentTestVars *testvars.TestVars + childTestVars *testvars.TestVars + + activeClusterIndex int + gates [2]*parentChildReplicationGate + removeHooks []func() + cleanups []func() + delayedTasks map[parentChildReplicationLane]*parentChildReplicationTask + metricCaptures [2]parentChildMetricCapture + trace []string + } + + parentChildMetricCapture struct { + handler *metricstest.CaptureHandler + capture *metricstest.Capture + } + + parentChildReplicationLane struct { + targetClusterIndex int + workflow parentChildWorkflow + } + + parentChildReplicationGate struct { + namespaceID string + workflowIDs map[string]struct{} + pending chan *parentChildReplicationTask + buffered map[string][]*parentChildReplicationTask + stop chan struct{} + stopOnce sync.Once + } + + parentChildReplicationTask struct { + task *replicationspb.ReplicationTask + metadata parentChildReplicationTaskMetadata + execute func() error + result chan error + + mu sync.Mutex + resolved bool + } + + parentChildReplicationTaskMetadata struct { + namespaceID string + workflowID string + runID string + } +) + +const ( + parentWorkflow parentChildWorkflow = iota + childWorkflow +) + +const ( + initialActiveCluster parentChildCluster = iota + initialStandbyCluster +) + +const ( + applyReplicationTask parentChildReplicationTaskAction = iota + delayReplicationTaskApply + ackReplicationTaskWithoutApplying +) + +const ( + historyClientVerifyChildCompletion = "HistoryClientVerifyChildExecutionCompletionRecorded" + historyClientVerifyFirstWorkflowTask = "HistoryClientVerifyFirstWorkflowTaskScheduled" +) + +func (s *parentChildXDCTestSuite) runParentChildScenario(scenario parentChildScenario) { + runtime := newParentChildScenarioRuntime(s) + defer runtime.close() + defer func() { + if s.T().Failed() { + s.T().Logf("parent-child scenario trace:\n%s", strings.Join(runtime.trace, "\n")) + } + }() + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + err := runtime.initialize(ctx) + cancel() + s.Require().NoError(err) + + for index, step := range scenario.steps { + runtime.tracef("step %d: %s", index+1, step.name) + ctx, cancel = context.WithTimeout(context.Background(), testTimeout) + err = step.run(ctx, runtime) + cancel() + s.Require().NoError(err, step.name) + } + + for index, expectation := range scenario.expectations { + runtime.tracef("expectation %d: %s", index+1, expectation.name) + await.Require(context.Background(), s.T(), func(t *await.T) { + require.NoError(t, expectation.check(t.Context(), runtime)) + }, testTimeout, replicationCheckInterval) + } +} + +func newParentChildScenarioRuntime(s *parentChildXDCTestSuite) *parentChildScenarioRuntime { + return &parentChildScenarioRuntime{ + suite: s, + delayedTasks: make(map[parentChildReplicationLane]*parentChildReplicationTask), + } +} + +func (r *parentChildScenarioRuntime) initialize(ctx context.Context) error { + if len(r.suite.clusters) != len(r.gates) { + return fmt.Errorf("parent-child scenarios require exactly %d clusters, got %d", len(r.gates), len(r.suite.clusters)) + } + + r.namespace = r.suite.createGlobalNamespace() + nsResp, err := r.suite.clusters[0].FrontendClient().DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{ + Namespace: r.namespace, + }) + if err != nil { + return err + } + r.namespaceID = nsResp.GetNamespaceInfo().GetId() + if r.suite.numHistoryShards < 2 { + return fmt.Errorf("parent-child scenarios require at least 2 history shards, got %d", r.suite.numHistoryShards) + } + + var parentShardID int32 + var childShardID int32 + r.parentID, r.childID, parentShardID, childShardID = workflowIDsOnDifferentShards(r.namespaceID, r.suite.numHistoryShards) + r.tracef("topology: parent shard=%d, child shard=%d", parentShardID, childShardID) + + for clusterIndex, cluster := range r.suite.clusters { + gate := newParentChildReplicationGate(r.namespaceID, r.parentID, r.childID) + r.gates[clusterIndex] = gate + r.removeHooks = append(r.removeHooks, cluster.InjectHook( + r.suite.T(), + testhooks.NewHook(testhooks.HistoryReplicationTaskInterceptor, gate.intercept), + testhooks.GlobalScope, + )) + + metricsHandler, ok := cluster.Host().GetMetricsHandler().(*metricstest.CaptureHandler) + if !ok { + return fmt.Errorf("cluster %d metrics handler does not support capture", clusterIndex) + } + r.metricCaptures[clusterIndex] = parentChildMetricCapture{ + handler: metricsHandler, + capture: metricsHandler.StartCapture(), + } + } + + r.parentTestVars = testvars.New(r.suite.T()).WithTaskQueue("parent-child-xdc-parent-task-queue") + r.childTestVars = testvars.New(r.suite.T()).WithTaskQueue("parent-child-xdc-child-task-queue") + return nil +} + +func (r *parentChildScenarioRuntime) close() { + for _, gate := range r.gates { + if gate != nil { + gate.close() + } + } + for index := len(r.removeHooks) - 1; index >= 0; index-- { + r.removeHooks[index]() + } + for _, capture := range r.metricCaptures { + if capture.handler != nil && capture.capture != nil { + capture.handler.StopCapture(capture.capture) + } + } + for index := len(r.cleanups) - 1; index >= 0; index-- { + r.cleanups[index]() + } +} + +func useLegacyHistoryReplication() parentChildScenarioStep { + return parentChildScenarioStep{ + name: "use legacy history replication for this scenario", + run: func(_ context.Context, runtime *parentChildScenarioRuntime) error { + for _, cluster := range runtime.suite.clusters { + runtime.cleanups = append(runtime.cleanups, cluster.OverrideDynamicConfig( + runtime.suite.T(), + dynamicconfig.EnableTransitionHistory, + false, + )) + } + return nil + }, + } +} + +func setLocalParentVerificationGrace( + cluster parentChildCluster, + duration time.Duration, +) parentChildScenarioStep { + return parentChildScenarioStep{ + name: fmt.Sprintf("set local parent verification grace on %s to %s", cluster, duration), + run: func(_ context.Context, runtime *parentChildScenarioRuntime) error { + clusterIndex := int(cluster) + if clusterIndex < 0 || clusterIndex >= len(runtime.suite.clusters) { + return fmt.Errorf("unknown parent-child cluster %d", cluster) + } + runtime.cleanups = append(runtime.cleanups, runtime.suite.clusters[clusterIndex].OverrideDynamicConfig( + runtime.suite.T(), + dynamicconfig.MaxLocalParentWorkflowVerificationDuration, + duration, + )) + return nil + }, + } +} + +func setStandbyClusterDelay( + cluster parentChildCluster, + duration time.Duration, +) parentChildScenarioStep { + return parentChildScenarioStep{ + name: fmt.Sprintf("set standby cluster delay on %s to %s", cluster, duration), + run: func(_ context.Context, runtime *parentChildScenarioRuntime) error { + clusterIndex := int(cluster) + if clusterIndex < 0 || clusterIndex >= len(runtime.suite.clusters) { + return fmt.Errorf("unknown parent-child cluster %d", cluster) + } + runtime.cleanups = append(runtime.cleanups, runtime.suite.clusters[clusterIndex].OverrideDynamicConfig( + runtime.suite.T(), + dynamicconfig.StandbyClusterDelay, + duration, + )) + return nil + }, + } +} + +func setStandbyTaskDiscardDelay( + cluster parentChildCluster, + taskType enumsspb.TaskType, + duration time.Duration, +) parentChildScenarioStep { + return parentChildScenarioStep{ + name: fmt.Sprintf("set standby %s discard delay on %s to %s", taskType, cluster, duration), + run: func(_ context.Context, runtime *parentChildScenarioRuntime) error { + clusterIndex := int(cluster) + if clusterIndex < 0 || clusterIndex >= len(runtime.suite.clusters) { + return fmt.Errorf("unknown parent-child cluster %d", cluster) + } + runtime.cleanups = append(runtime.cleanups, runtime.suite.clusters[clusterIndex].OverrideDynamicConfig( + runtime.suite.T(), + dynamicconfig.StandbyTaskMissingEventsDiscardDelay, + []dynamicconfig.ConstrainedValue{{ + Constraints: dynamicconfig.Constraints{TaskType: taskType}, + Value: duration, + }}, + )) + return nil + }, + } +} + +func startParentWorkflow() parentChildScenarioStep { + return parentChildScenarioStep{ + name: "start parent workflow on the active cluster", + run: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.startParentWorkflow(ctx) + }, + } +} + +// Event checkpoints identify a replication task; each action applies to the entire task. +func applyReplicationThroughTaskContainingEvent( + targetCluster parentChildCluster, + workflow parentChildWorkflow, + eventType enumspb.EventType, +) parentChildScenarioStep { + return replicationTaskStep(applyReplicationTask, targetCluster, workflow, eventType) +} + +func delayReplicationAtTaskContainingEvent( + targetCluster parentChildCluster, + workflow parentChildWorkflow, + eventType enumspb.EventType, +) parentChildScenarioStep { + return replicationTaskStep(delayReplicationTaskApply, targetCluster, workflow, eventType) +} + +func applyDelayedReplication( + targetCluster parentChildCluster, + workflow parentChildWorkflow, +) parentChildScenarioStep { + return parentChildScenarioStep{ + name: fmt.Sprintf("apply delayed %s replication to %s", workflow, targetCluster), + run: func(_ context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.applyDelayedReplication(targetCluster, workflow) + }, + } +} + +func acknowledgeReplicationTaskContainingEventWithoutApplying( + targetCluster parentChildCluster, + workflow parentChildWorkflow, + eventType enumspb.EventType, +) parentChildScenarioStep { + return replicationTaskStep(ackReplicationTaskWithoutApplying, targetCluster, workflow, eventType) +} + +func replicationTaskStep( + action parentChildReplicationTaskAction, + targetCluster parentChildCluster, + workflow parentChildWorkflow, + eventType enumspb.EventType, +) parentChildScenarioStep { + return parentChildScenarioStep{ + name: fmt.Sprintf("%s %s replication to %s at task containing %s", action, workflow, targetCluster, eventType), + run: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.processReplicationThroughTaskContainingEvent(ctx, targetCluster, workflow, eventType, action) + }, + } +} + +func completeParentWorkflowTaskWithStartChildCommand() parentChildScenarioStep { + return parentChildScenarioStep{ + name: "complete parent workflow task with a StartChild command", + run: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.completeParentWorkflowTaskWithStartChildCommand(ctx) + }, + } +} + +func completeChildWorkflowTask() parentChildScenarioStep { + return parentChildScenarioStep{ + name: "complete the child workflow task", + run: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.completeChildWorkflowTask(ctx) + }, + } +} + +func waitForWorkflowEventOnCluster( + cluster parentChildCluster, + workflow parentChildWorkflow, + eventType enumspb.EventType, +) parentChildScenarioStep { + expectation := workflowHasEventOnCluster(cluster, workflow, eventType) + return parentChildScenarioStep{ + name: "wait until " + expectation.name, + run: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.waitForExpectation(ctx, expectation) + }, + } +} + +func confirmWorkflowIsMissingOnCluster( + cluster parentChildCluster, + workflow parentChildWorkflow, +) parentChildScenarioStep { + return parentChildScenarioStep{ + name: fmt.Sprintf("confirm %s is missing on %s", workflow, cluster), + run: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.confirmWorkflowMissing(ctx, cluster, workflow) + }, + } +} + +func forceFailoverNamespaceTo(targetCluster parentChildCluster) parentChildScenarioStep { + return parentChildScenarioStep{ + name: fmt.Sprintf("force fail over the namespace to %s", targetCluster), + run: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.forceFailover(ctx, targetCluster) + }, + } +} + +func currentWorkflowHasStatusOnCluster( + cluster parentChildCluster, + workflow parentChildWorkflow, + status enumspb.WorkflowExecutionStatus, +) parentChildExpectation { + return parentChildExpectation{ + name: fmt.Sprintf("current %s has status %s on %s", workflow, status, cluster), + check: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + clusterIndex := int(cluster) + if clusterIndex < 0 || clusterIndex >= len(runtime.suite.clusters) { + return fmt.Errorf("unknown parent-child cluster %d", cluster) + } + workflowID, err := runtime.workflowID(workflow) + if err != nil { + return err + } + resp, err := runtime.suite.clusters[clusterIndex].FrontendClient().DescribeWorkflowExecution( + ctx, + &workflowservice.DescribeWorkflowExecutionRequest{ + Namespace: runtime.namespace, + Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID}, + }, + ) + if err != nil { + return err + } + info := resp.GetWorkflowExecutionInfo() + if runID := info.GetExecution().GetRunId(); runID != runtime.workflowRunID(workflow) { + return fmt.Errorf("%s run is %q, want %q", workflow, runID, runtime.workflowRunID(workflow)) + } + if actualStatus := info.GetStatus(); actualStatus != status { + return fmt.Errorf("%s status is %s, want %s", workflow, actualStatus, status) + } + return nil + }, + } +} + +func workflowIsMissingOnCluster( + cluster parentChildCluster, + workflow parentChildWorkflow, +) parentChildExpectation { + return parentChildExpectation{ + name: fmt.Sprintf("%s is missing on %s", workflow, cluster), + check: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.confirmWorkflowMissing(ctx, cluster, workflow) + }, + } +} + +func workflowHasEventOnCluster( + cluster parentChildCluster, + workflow parentChildWorkflow, + eventType enumspb.EventType, +) parentChildExpectation { + return parentChildExpectation{ + name: fmt.Sprintf("%s contains %s on %s", workflow, eventType, cluster), + check: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + events, err := runtime.workflowHistoryOnCluster(ctx, cluster, workflow) + if err != nil { + return err + } + if !historyContainsEvent(events, eventType) { + return fmt.Errorf("%s has no %s on %s", workflow, eventType, cluster) + } + return nil + }, + } +} + +func historyVerificationFailedOnCluster( + cluster parentChildCluster, + operation string, + expectedError error, +) parentChildExpectation { + errorType := metrics.ServiceErrorTypeTag(expectedError).Value + return parentChildExpectation{ + name: fmt.Sprintf("%s fails with %s on %s", operation, errorType, cluster), + check: func(_ context.Context, runtime *parentChildScenarioRuntime) error { + namespaceTag := metrics.NamespaceTag(runtime.namespace) + serviceRoleTag := metrics.ServiceRoleTag(metrics.HistoryRoleTagValue) + return runtime.requireCapturedMetric(cluster, metrics.ClientFailures.Name(), map[string]string{ + metrics.OperationTagName: operation, + metrics.ErrorTypeTagName: errorType, + namespaceTag.Key: namespaceTag.Value, + serviceRoleTag.Key: serviceRoleTag.Value, + }) + }, + } +} + +func waitForHistoryVerificationFailureOnCluster( + cluster parentChildCluster, + operation string, + expectedError error, +) parentChildScenarioStep { + expectation := historyVerificationFailedOnCluster(cluster, operation, expectedError) + return parentChildScenarioStep{ + name: "wait until " + expectation.name, + run: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.waitForExpectation(ctx, expectation) + }, + } +} + +func taskWasDiscardedOnCluster( + cluster parentChildCluster, + taskType string, +) parentChildExpectation { + return parentChildExpectation{ + name: fmt.Sprintf("%s is discarded on %s", taskType, cluster), + check: func(_ context.Context, runtime *parentChildScenarioRuntime) error { + namespaceTag := metrics.NamespaceTag(runtime.namespace) + return runtime.requireCapturedMetric(cluster, metrics.TaskDiscarded.Name(), map[string]string{ + metrics.OperationTagName: taskType, + metrics.TaskTypeTagName: taskType, + namespaceTag.Key: namespaceTag.Value, + }) + }, + } +} + +func (r *parentChildScenarioRuntime) startParentWorkflow(ctx context.Context) error { + if r.parentRunID != "" { + return fmt.Errorf("parent workflow is already started with run ID %q", r.parentRunID) + } + startResp, err := r.activeCluster().FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{ + Namespace: r.namespace, + WorkflowId: r.parentID, + WorkflowType: &commonpb.WorkflowType{Name: "parent-workflow"}, + TaskQueue: r.parentTestVars.TaskQueue(), + RequestId: uuid.NewString(), + WorkflowRunTimeout: durationpb.New(time.Minute), + WorkflowTaskTimeout: durationpb.New(10 * time.Second), + Identity: r.parentTestVars.WorkerIdentity(), + }) + if err != nil { + return err + } + r.parentRunID = startResp.GetRunId() + r.tracef(" started parent %s/%s on cluster %d", r.parentID, r.parentRunID, r.activeClusterIndex) + return nil +} + +func (r *parentChildScenarioRuntime) processReplicationThroughTaskContainingEvent( + ctx context.Context, + targetCluster parentChildCluster, + workflow parentChildWorkflow, + eventType enumspb.EventType, + targetAction parentChildReplicationTaskAction, +) error { + targetClusterIndex := int(targetCluster) + if targetClusterIndex < 0 || targetClusterIndex >= len(r.gates) { + return fmt.Errorf("unknown parent-child cluster %d", targetCluster) + } + lane := parentChildReplicationLane{targetClusterIndex: targetClusterIndex, workflow: workflow} + workflowID, err := r.workflowID(workflow) + if err != nil { + return err + } + + for { + task, delayed := r.delayedTasks[lane] + if !delayed { + task, err = r.gates[targetClusterIndex].nextForWorkflow(ctx, workflowID) + if err != nil { + return err + } + } + + events, err := decodeParentChildReplicationEvents(task.task) + if err != nil { + return err + } + containsCheckpoint := historyContainsEvent(events, eventType) + action := applyReplicationTask + if containsCheckpoint { + action = targetAction + } + + r.tracef( + " %s task %d to cluster %d for %s [%s]", + action, + task.task.GetSourceTaskId(), + targetClusterIndex, + workflow, + formatParentChildReplicationTask(task.task, events), + ) + switch action { + case applyReplicationTask: + delete(r.delayedTasks, lane) + if err := task.apply(); err != nil { + return err + } + r.recordChildRunIDFromAppliedTask(workflow, task, events) + case delayReplicationTaskApply: + r.delayedTasks[lane] = task + case ackReplicationTaskWithoutApplying: + delete(r.delayedTasks, lane) + if err := task.acknowledgeWithoutApplying(); err != nil { + return err + } + default: + return fmt.Errorf("unknown replication task action %d", action) + } + + if containsCheckpoint { + return nil + } + } +} + +func (r *parentChildScenarioRuntime) applyDelayedReplication( + targetCluster parentChildCluster, + workflow parentChildWorkflow, +) error { + targetClusterIndex := int(targetCluster) + if targetClusterIndex < 0 || targetClusterIndex >= len(r.gates) { + return fmt.Errorf("unknown parent-child cluster %d", targetCluster) + } + if _, err := r.workflowID(workflow); err != nil { + return err + } + + lane := parentChildReplicationLane{targetClusterIndex: targetClusterIndex, workflow: workflow} + task, delayed := r.delayedTasks[lane] + if !delayed { + return fmt.Errorf("no delayed %s replication task to %s", workflow, targetCluster) + } + events, err := decodeParentChildReplicationEvents(task.task) + if err != nil { + return err + } + + r.tracef( + " apply delayed task %d to cluster %d for %s [%s]", + task.task.GetSourceTaskId(), + targetClusterIndex, + workflow, + formatParentChildReplicationTask(task.task, events), + ) + delete(r.delayedTasks, lane) + if err := task.apply(); err != nil { + return err + } + r.recordChildRunIDFromAppliedTask(workflow, task, events) + return nil +} + +func (r *parentChildScenarioRuntime) completeParentWorkflowTaskWithStartChildCommand(ctx context.Context) error { + if r.parentRunID == "" { + return errors.New("parent workflow is not started") + } + poller := taskpoller.New(r.suite.T(), r.activeCluster().FrontendClient(), r.namespace) + _, err := poller.PollAndHandleWorkflowTask(r.parentTestVars, func( + task *workflowservice.PollWorkflowTaskQueueResponse, + ) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) { + if task.GetWorkflowExecution().GetWorkflowId() != r.parentID || task.GetWorkflowExecution().GetRunId() != r.parentRunID { + return nil, fmt.Errorf( + "polled workflow %s/%s, want parent %s/%s", + task.GetWorkflowExecution().GetWorkflowId(), + task.GetWorkflowExecution().GetRunId(), + r.parentID, + r.parentRunID, + ) + } + return &workflowservice.RespondWorkflowTaskCompletedRequest{ + Commands: []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_StartChildWorkflowExecutionCommandAttributes{ + StartChildWorkflowExecutionCommandAttributes: &commandpb.StartChildWorkflowExecutionCommandAttributes{ + WorkflowId: r.childID, + WorkflowType: &commonpb.WorkflowType{Name: "child-workflow"}, + TaskQueue: r.childTestVars.TaskQueue(), + WorkflowRunTimeout: durationpb.New(time.Minute), + WorkflowTaskTimeout: durationpb.New(10 * time.Second), + ParentClosePolicy: enumspb.PARENT_CLOSE_POLICY_ABANDON, + }, + }, + }}, + }, nil + }, taskpoller.WithTimeout(testTimeout)) + return err +} + +func (r *parentChildScenarioRuntime) completeChildWorkflowTask(ctx context.Context) error { + poller := taskpoller.New(r.suite.T(), r.activeCluster().FrontendClient(), r.namespace) + _, err := poller.PollAndHandleWorkflowTask(r.childTestVars, func( + task *workflowservice.PollWorkflowTaskQueueResponse, + ) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) { + execution := task.GetWorkflowExecution() + if execution.GetWorkflowId() != r.childID { + return nil, fmt.Errorf("polled workflow %s/%s, want child %s", execution.GetWorkflowId(), execution.GetRunId(), r.childID) + } + if r.childRunID == "" { + r.childRunID = execution.GetRunId() + } + if execution.GetRunId() != r.childRunID { + return nil, fmt.Errorf("polled child run %s, want %s", execution.GetRunId(), r.childRunID) + } + return &workflowservice.RespondWorkflowTaskCompletedRequest{ + Commands: []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION, + Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{ + CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{}, + }, + }}, + }, nil + }, taskpoller.WithTimeout(testTimeout)) + return err +} + +func (r *parentChildScenarioRuntime) workflowHistoryOnCluster( + ctx context.Context, + cluster parentChildCluster, + workflow parentChildWorkflow, +) ([]*historypb.HistoryEvent, error) { + clusterIndex := int(cluster) + if clusterIndex < 0 || clusterIndex >= len(r.suite.clusters) { + return nil, fmt.Errorf("unknown parent-child cluster %d", cluster) + } + workflowID, err := r.workflowID(workflow) + if err != nil { + return nil, err + } + runID := r.workflowRunID(workflow) + resp, err := r.suite.clusters[clusterIndex].FrontendClient().GetWorkflowExecutionHistory(ctx, &workflowservice.GetWorkflowExecutionHistoryRequest{ + Namespace: r.namespace, + Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: runID}, + }) + if err != nil { + return nil, err + } + return resp.GetHistory().GetEvents(), nil +} + +func (r *parentChildScenarioRuntime) workflowMutableState( + ctx context.Context, + cluster parentChildCluster, + workflow parentChildWorkflow, +) (*persistencespb.WorkflowMutableState, error) { + clusterIndex := int(cluster) + if clusterIndex < 0 || clusterIndex >= len(r.suite.clusters) { + return nil, fmt.Errorf("unknown parent-child cluster %d", cluster) + } + workflowID, err := r.workflowID(workflow) + if err != nil { + return nil, err + } + resp, err := r.suite.clusters[clusterIndex].HistoryClient().DescribeMutableState(ctx, &historyservice.DescribeMutableStateRequest{ + NamespaceId: r.namespaceID, + Execution: &commonpb.WorkflowExecution{ + WorkflowId: workflowID, + RunId: r.workflowRunID(workflow), + }, + ArchetypeId: chasm.WorkflowArchetypeID, + }) + if err != nil { + return nil, err + } + if resp.GetDatabaseMutableState() == nil { + return nil, fmt.Errorf("%s database mutable state is nil on %s", workflow, cluster) + } + return resp.GetDatabaseMutableState(), nil +} + +func (r *parentChildScenarioRuntime) confirmWorkflowMissing( + ctx context.Context, + cluster parentChildCluster, + workflow parentChildWorkflow, +) error { + _, err := r.workflowMutableState(ctx, cluster, workflow) + var notFound *serviceerror.NotFound + if errors.As(err, ¬Found) { + return nil + } + if err == nil { + return fmt.Errorf("%s unexpectedly exists on %s", workflow, cluster) + } + return err +} + +func (r *parentChildScenarioRuntime) requireCapturedMetric( + cluster parentChildCluster, + metricName string, + tags map[string]string, +) error { + clusterIndex := int(cluster) + if clusterIndex < 0 || clusterIndex >= len(r.metricCaptures) { + return fmt.Errorf("unknown parent-child cluster %d", cluster) + } + capture := r.metricCaptures[clusterIndex].capture + if capture == nil { + return fmt.Errorf("metrics capture is not initialized for %s", cluster) + } + recordings := capture.Snapshot()[metricName] + for _, recording := range recordings { + matches := true + for key, value := range tags { + if recording.Tags[key] != value { + matches = false + break + } + } + if matches { + return nil + } + } + recordedTags := make([]map[string]string, 0, len(recordings)) + for _, recording := range recordings { + recordedTags = append(recordedTags, recording.Tags) + } + return fmt.Errorf("metric %q with tags %v was not captured on %s; recorded tags: %v", metricName, tags, cluster, recordedTags) +} + +func (r *parentChildScenarioRuntime) waitForExpectation( + ctx context.Context, + expectation parentChildExpectation, +) error { + ticker := time.NewTicker(replicationCheckInterval) + defer ticker.Stop() + for { + err := expectation.check(ctx, r) + if err == nil { + return nil + } + select { + case <-ctx.Done(): + return fmt.Errorf("waiting for %s: %w; last observation: %v", expectation.name, ctx.Err(), err) + case <-ticker.C: + } + } +} + +func (r *parentChildScenarioRuntime) forceFailover(ctx context.Context, target parentChildCluster) error { + targetClusterIndex := int(target) + if targetClusterIndex < 0 || targetClusterIndex >= len(r.suite.clusters) { + return fmt.Errorf("unknown parent-child cluster %d", target) + } + if targetClusterIndex == r.activeClusterIndex { + return fmt.Errorf("cluster %s is already active", target) + } + targetCluster := r.suite.clusters[targetClusterIndex].ClusterName() + _, err := r.activeCluster().FrontendClient().UpdateNamespace(ctx, &workflowservice.UpdateNamespaceRequest{ + Namespace: r.namespace, + ReplicationConfig: &replicationpb.NamespaceReplicationConfig{ + ActiveClusterName: targetCluster, + }, + }) + if err != nil { + return err + } + + await.Require(ctx, r.suite.T(), func(t *await.T) { + for _, cluster := range r.suite.clusters { + resp, describeErr := cluster.FrontendClient().DescribeNamespace(t.Context(), &workflowservice.DescribeNamespaceRequest{ + Namespace: r.namespace, + }) + require.NoError(t, describeErr) + require.Equal(t, targetCluster, resp.GetReplicationConfig().GetActiveClusterName()) + } + }, replicationWaitTime, replicationCheckInterval) + r.suite.waitForNamespaceCacheRefresh() + r.activeClusterIndex = targetClusterIndex + r.tracef(" active cluster is now %d (%s)", targetClusterIndex, targetCluster) + return nil +} + +func (r *parentChildScenarioRuntime) activeWorkflowHistory( + ctx context.Context, + workflow parentChildWorkflow, +) ([]*historypb.HistoryEvent, error) { + return r.workflowHistoryOnCluster(ctx, parentChildCluster(r.activeClusterIndex), workflow) +} + +func (r *parentChildScenarioRuntime) activeCluster() *testcore.TestCluster { + return r.suite.clusters[r.activeClusterIndex] +} + +func (r *parentChildScenarioRuntime) workflowID(workflow parentChildWorkflow) (string, error) { + switch workflow { + case parentWorkflow: + return r.parentID, nil + case childWorkflow: + return r.childID, nil + default: + return "", fmt.Errorf("unknown parent-child workflow %d", workflow) + } +} + +func (r *parentChildScenarioRuntime) workflowRunID(workflow parentChildWorkflow) string { + switch workflow { + case parentWorkflow: + return r.parentRunID + case childWorkflow: + return r.childRunID + default: + return "" + } +} + +func (r *parentChildScenarioRuntime) recordChildRunIDFromAppliedTask( + workflow parentChildWorkflow, + task *parentChildReplicationTask, + events []*historypb.HistoryEvent, +) { + if workflow == childWorkflow && historyContainsEvent(events, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED) { + r.childRunID = task.metadata.runID + } +} + +func (r *parentChildScenarioRuntime) tracef(format string, args ...any) { + r.trace = append(r.trace, fmt.Sprintf(format, args...)) +} + +func newParentChildReplicationGate(namespaceID string, workflowIDs ...string) *parentChildReplicationGate { + gate := &parentChildReplicationGate{ + namespaceID: namespaceID, + workflowIDs: make(map[string]struct{}, len(workflowIDs)), + pending: make(chan *parentChildReplicationTask, 32), + buffered: make(map[string][]*parentChildReplicationTask), + stop: make(chan struct{}), + } + for _, workflowID := range workflowIDs { + gate.workflowIDs[workflowID] = struct{}{} + } + return gate +} + +func (g *parentChildReplicationGate) intercept( + task *replicationspb.ReplicationTask, + execute func() error, +) error { + metadata, ok := getParentChildReplicationTaskMetadata(task) + if !ok || metadata.namespaceID != g.namespaceID { + return execute() + } + if _, workflowSelected := g.workflowIDs[metadata.workflowID]; !workflowSelected { + return execute() + } + + bufferedTask := &parentChildReplicationTask{ + task: task, + metadata: metadata, + execute: execute, + result: make(chan error, 1), + } + select { + case g.pending <- bufferedTask: + case <-g.stop: + return nil + } + select { + case err := <-bufferedTask.result: + return err + case <-g.stop: + return nil + } +} + +func (g *parentChildReplicationGate) nextForWorkflow( + ctx context.Context, + workflowID string, +) (*parentChildReplicationTask, error) { + select { + case <-g.stop: + return nil, errParentChildReplicationGateClosed + default: + } + + if tasks := g.buffered[workflowID]; len(tasks) > 0 { + task := tasks[0] + g.buffered[workflowID] = tasks[1:] + return task, nil + } + + for { + select { + case task := <-g.pending: + taskWorkflowID := task.metadata.workflowID + if taskWorkflowID == workflowID { + return task, nil + } + g.buffered[taskWorkflowID] = append(g.buffered[taskWorkflowID], task) + case <-g.stop: + return nil, errParentChildReplicationGateClosed + case <-ctx.Done(): + return nil, ctx.Err() + } + } +} + +func (g *parentChildReplicationGate) close() { + g.stopOnce.Do(func() { + close(g.stop) + }) +} + +func (t *parentChildReplicationTask) apply() error { + return t.resolve(true) +} + +func (t *parentChildReplicationTask) acknowledgeWithoutApplying() error { + return t.resolve(false) +} + +func (t *parentChildReplicationTask) resolve(shouldApply bool) error { + t.mu.Lock() + if t.resolved { + t.mu.Unlock() + return errParentChildTaskAlreadyResolved + } + t.resolved = true + t.mu.Unlock() + + var err error + if shouldApply { + err = t.execute() + } + t.result <- err + return err +} + +func getParentChildReplicationTaskMetadata( + task *replicationspb.ReplicationTask, +) (parentChildReplicationTaskMetadata, bool) { + if task == nil { + return parentChildReplicationTaskMetadata{}, false + } + if info := task.GetRawTaskInfo(); info != nil { + return parentChildReplicationTaskMetadata{ + namespaceID: info.GetNamespaceId(), + workflowID: info.GetWorkflowId(), + runID: info.GetRunId(), + }, true + } + + var metadata parentChildReplicationTaskMetadata + switch attrs := task.GetAttributes().(type) { + case *replicationspb.ReplicationTask_SyncWorkflowStateTaskAttributes: + state := attrs.SyncWorkflowStateTaskAttributes.GetWorkflowState() + metadata.namespaceID = state.GetExecutionInfo().GetNamespaceId() + metadata.workflowID = state.GetExecutionInfo().GetWorkflowId() + metadata.runID = state.GetExecutionState().GetRunId() + case *replicationspb.ReplicationTask_SyncActivityTaskAttributes: + metadata.namespaceID = attrs.SyncActivityTaskAttributes.GetNamespaceId() + metadata.workflowID = attrs.SyncActivityTaskAttributes.GetWorkflowId() + metadata.runID = attrs.SyncActivityTaskAttributes.GetRunId() + case *replicationspb.ReplicationTask_HistoryTaskAttributes: + metadata.namespaceID = attrs.HistoryTaskAttributes.GetNamespaceId() + metadata.workflowID = attrs.HistoryTaskAttributes.GetWorkflowId() + metadata.runID = attrs.HistoryTaskAttributes.GetRunId() + case *replicationspb.ReplicationTask_SyncHsmAttributes: + metadata.namespaceID = attrs.SyncHsmAttributes.GetNamespaceId() + metadata.workflowID = attrs.SyncHsmAttributes.GetWorkflowId() + metadata.runID = attrs.SyncHsmAttributes.GetRunId() + case *replicationspb.ReplicationTask_BackfillHistoryTaskAttributes: + metadata.namespaceID = attrs.BackfillHistoryTaskAttributes.GetNamespaceId() + metadata.workflowID = attrs.BackfillHistoryTaskAttributes.GetWorkflowId() + metadata.runID = attrs.BackfillHistoryTaskAttributes.GetRunId() + case *replicationspb.ReplicationTask_VerifyVersionedTransitionTaskAttributes: + metadata.namespaceID = attrs.VerifyVersionedTransitionTaskAttributes.GetNamespaceId() + metadata.workflowID = attrs.VerifyVersionedTransitionTaskAttributes.GetWorkflowId() + metadata.runID = attrs.VerifyVersionedTransitionTaskAttributes.GetRunId() + case *replicationspb.ReplicationTask_SyncVersionedTransitionTaskAttributes: + metadata.namespaceID = attrs.SyncVersionedTransitionTaskAttributes.GetNamespaceId() + metadata.workflowID = attrs.SyncVersionedTransitionTaskAttributes.GetWorkflowId() + metadata.runID = attrs.SyncVersionedTransitionTaskAttributes.GetRunId() + default: + return parentChildReplicationTaskMetadata{}, false + } + return metadata, true +} + +func decodeParentChildReplicationEvents(task *replicationspb.ReplicationTask) ([]*historypb.HistoryEvent, error) { + var blobs []*commonpb.DataBlob + switch attrs := task.GetAttributes().(type) { + case *replicationspb.ReplicationTask_HistoryTaskAttributes: + blobs = attrs.HistoryTaskAttributes.GetEventsBatches() + if attrs.HistoryTaskAttributes.GetEvents() != nil { + blobs = []*commonpb.DataBlob{attrs.HistoryTaskAttributes.GetEvents()} + } + if len(blobs) == 0 { + return nil, errors.New("history replication task has no event batch") + } + case *replicationspb.ReplicationTask_SyncVersionedTransitionTaskAttributes: + artifact := attrs.SyncVersionedTransitionTaskAttributes.GetVersionedTransitionArtifact() + if artifact == nil { + return nil, errors.New("sync versioned transition task has no artifact") + } + blobs = artifact.GetEventBatches() + case *replicationspb.ReplicationTask_BackfillHistoryTaskAttributes: + blobs = attrs.BackfillHistoryTaskAttributes.GetEventBatches() + if len(blobs) == 0 { + return nil, errors.New("backfill history replication task has no event batch") + } + case *replicationspb.ReplicationTask_VerifyVersionedTransitionTaskAttributes, + *replicationspb.ReplicationTask_SyncWorkflowStateTaskAttributes, + *replicationspb.ReplicationTask_SyncActivityTaskAttributes, + *replicationspb.ReplicationTask_SyncHsmAttributes: + return nil, nil + default: + if _, ok := getParentChildReplicationTaskMetadata(task); ok { + return nil, nil + } + return nil, errors.New("replication task has no workflow metadata") + } + + var events []*historypb.HistoryEvent + for _, blob := range blobs { + batch, err := serialization.DefaultDecoder.DeserializeEvents(blob) + if err != nil { + return nil, err + } + events = append(events, batch...) + } + return events, nil +} + +func historyContainsEvent(events []*historypb.HistoryEvent, eventType enumspb.EventType) bool { + for _, event := range events { + if event.GetEventType() == eventType { + return true + } + } + return false +} + +func findHistoryEvent( + events []*historypb.HistoryEvent, + eventType enumspb.EventType, + matches func(*historypb.HistoryEvent) bool, +) *historypb.HistoryEvent { + for _, event := range events { + if event.GetEventType() == eventType && (matches == nil || matches(event)) { + return event + } + } + return nil +} + +func formatParentChildEventTypes(events []*historypb.HistoryEvent) string { + eventTypes := make([]string, 0, len(events)) + for _, event := range events { + eventTypes = append(eventTypes, strings.TrimPrefix(event.GetEventType().String(), "EVENT_TYPE_")) + } + return strings.Join(eventTypes, ", ") +} + +func formatParentChildReplicationTask( + task *replicationspb.ReplicationTask, + events []*historypb.HistoryEvent, +) string { + if len(events) != 0 { + return formatParentChildEventTypes(events) + } + switch task.GetAttributes().(type) { + case *replicationspb.ReplicationTask_SyncWorkflowStateTaskAttributes: + return "SyncWorkflowState" + case *replicationspb.ReplicationTask_SyncActivityTaskAttributes: + return "SyncActivity" + case *replicationspb.ReplicationTask_SyncHsmAttributes: + return "SyncHSM" + case *replicationspb.ReplicationTask_VerifyVersionedTransitionTaskAttributes: + return "VerifyVersionedTransition" + case *replicationspb.ReplicationTask_SyncVersionedTransitionTaskAttributes: + return "SyncVersionedTransition" + default: + return task.GetTaskType().String() + } +} + +func workflowIDsOnDifferentShards( + namespaceID string, + numHistoryShards int32, +) (parentID string, childID string, parentShardID int32, childShardID int32) { + parentID = "parent-" + uuid.NewString() + parentShardID = common.WorkflowIDToHistoryShard(namespaceID, parentID, numHistoryShards) + for { + childID = "child-" + uuid.NewString() + childShardID = common.WorkflowIDToHistoryShard(namespaceID, childID, numHistoryShards) + if childShardID != parentShardID { + return parentID, childID, parentShardID, childShardID + } + } +} + +func (workflow parentChildWorkflow) String() string { + switch workflow { + case parentWorkflow: + return "parent" + case childWorkflow: + return "child" + default: + return fmt.Sprintf("workflow(%d)", workflow) + } +} + +func (cluster parentChildCluster) String() string { + switch cluster { + case initialActiveCluster: + return "initial active cluster" + case initialStandbyCluster: + return "initial standby cluster" + default: + return fmt.Sprintf("cluster(%d)", cluster) + } +} + +func (action parentChildReplicationTaskAction) String() string { + switch action { + case applyReplicationTask: + return "apply" + case delayReplicationTaskApply: + return "delay-apply" + case ackReplicationTaskWithoutApplying: + return "ack-without-apply" + default: + return fmt.Sprintf("action(%d)", action) + } +} diff --git a/tests/xdc/parent_child_harness_test.go b/tests/xdc/parent_child_harness_test.go new file mode 100644 index 0000000000..c787a243d4 --- /dev/null +++ b/tests/xdc/parent_child_harness_test.go @@ -0,0 +1,398 @@ +package xdc + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + enumsspb "go.temporal.io/server/api/enums/v1" + replicationspb "go.temporal.io/server/api/replication/v1" + "go.temporal.io/server/common/persistence/serialization" + "go.temporal.io/server/common/testing/await" + "go.temporal.io/server/common/testing/protorequire" +) + +func TestParentChildReplicationGate(t *testing.T) { + t.Run("passes unrelated tasks through", func(t *testing.T) { + gate := newParentChildReplicationGate("namespace", "parent", "child") + defer gate.close() + + executeErr := errors.New("execute failed") + var executions atomic.Int32 + err := gate.intercept( + newParentChildHistoryReplicationTask("other-namespace", "parent", 1), + func() error { + executions.Add(1) + return executeErr + }, + ) + require.ErrorIs(t, err, executeErr) + require.Equal(t, int32(1), executions.Load()) + + err = gate.intercept( + newParentChildHistoryReplicationTask("namespace", "other-workflow", 2), + func() error { + executions.Add(1) + return executeErr + }, + ) + require.ErrorIs(t, err, executeErr) + require.Equal(t, int32(2), executions.Load()) + }) + + t.Run("applies and acknowledges without applying exactly once", func(t *testing.T) { + gate := newParentChildReplicationGate("namespace", "parent", "child") + defer gate.close() + + applyErr := errors.New("apply failed") + var applied atomic.Int32 + applyResult := interceptParentChildTask( + gate, + newParentChildHistoryReplicationTask("namespace", "parent", 1), + func() error { + applied.Add(1) + return applyErr + }, + ) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + deferredTask, err := gate.nextForWorkflow(ctx, "parent") + cancel() + require.NoError(t, err) + require.Zero(t, applied.Load()) + require.ErrorIs(t, deferredTask.apply(), applyErr) + require.ErrorIs(t, receiveParentChildInterceptResult(t, applyResult), applyErr) + require.ErrorIs(t, deferredTask.apply(), errParentChildTaskAlreadyResolved) + require.Equal(t, int32(1), applied.Load()) + + var appliedWithoutPermission atomic.Int32 + ackResult := interceptParentChildTask( + gate, + newParentChildHistoryReplicationTask("namespace", "child", 2), + func() error { + appliedWithoutPermission.Add(1) + return nil + }, + ) + ctx, cancel = context.WithTimeout(context.Background(), time.Second) + deferredTask, err = gate.nextForWorkflow(ctx, "child") + cancel() + require.NoError(t, err) + require.NoError(t, deferredTask.acknowledgeWithoutApplying()) + require.NoError(t, receiveParentChildInterceptResult(t, ackResult)) + require.ErrorIs(t, deferredTask.acknowledgeWithoutApplying(), errParentChildTaskAlreadyResolved) + require.Zero(t, appliedWithoutPermission.Load()) + }) + + t.Run("keeps tasks for the other workflow buffered", func(t *testing.T) { + gate := newParentChildReplicationGate("namespace", "parent", "child") + defer gate.close() + + childResult := interceptParentChildTask( + gate, + newParentChildHistoryReplicationTask("namespace", "child", 1), + func() error { return nil }, + ) + await.RequireTrue(t, func() bool { return len(gate.pending) == 1 }, time.Second, time.Millisecond) + parentResult := interceptParentChildTask( + gate, + newParentChildHistoryReplicationTask("namespace", "parent", 2), + func() error { return nil }, + ) + await.RequireTrue(t, func() bool { return len(gate.pending) == 2 }, time.Second, time.Millisecond) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + parentTask, err := gate.nextForWorkflow(ctx, "parent") + cancel() + require.NoError(t, err) + require.Equal(t, int64(2), parentTask.task.GetSourceTaskId()) + require.Len(t, gate.buffered["child"], 1) + require.NoError(t, parentTask.acknowledgeWithoutApplying()) + require.NoError(t, receiveParentChildInterceptResult(t, parentResult)) + + ctx, cancel = context.WithTimeout(context.Background(), time.Second) + childTask, err := gate.nextForWorkflow(ctx, "child") + cancel() + require.NoError(t, err) + require.Equal(t, int64(1), childTask.task.GetSourceTaskId()) + require.NoError(t, childTask.acknowledgeWithoutApplying()) + require.NoError(t, receiveParentChildInterceptResult(t, childResult)) + }) + + t.Run("close releases delayed and queued tasks", func(t *testing.T) { + gate := newParentChildReplicationGate("namespace", "parent", "child") + delayedResult := interceptParentChildTask( + gate, + newParentChildHistoryReplicationTask("namespace", "parent", 1), + func() error { return nil }, + ) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + _, err := gate.nextForWorkflow(ctx, "parent") + cancel() + require.NoError(t, err) + + queuedResult := interceptParentChildTask( + gate, + newParentChildHistoryReplicationTask("namespace", "child", 2), + func() error { return nil }, + ) + await.RequireTrue(t, func() bool { return len(gate.pending) == 1 }, time.Second, time.Millisecond) + + gate.close() + gate.close() + require.NoError(t, receiveParentChildInterceptResult(t, delayedResult)) + require.NoError(t, receiveParentChildInterceptResult(t, queuedResult)) + + ctx, cancel = context.WithTimeout(context.Background(), time.Second) + _, err = gate.nextForWorkflow(ctx, "child") + cancel() + require.ErrorIs(t, err, errParentChildReplicationGateClosed) + }) +} + +func TestParentChildHarnessAppliesPreviouslyDelayedTaskAfterActiveClusterChanges(t *testing.T) { + gate := newParentChildReplicationGate("namespace", "parent", "child") + defer gate.close() + + events := []*historypb.HistoryEvent{{ + EventId: 3, + EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED, + }} + blob, err := serialization.NewSerializer().SerializeEvents(events) + require.NoError(t, err) + task := newParentChildHistoryReplicationTask("namespace", "parent", 1) + task.GetHistoryTaskAttributes().Events = blob + + var executions atomic.Int32 + interceptResult := interceptParentChildTask(gate, task, func() error { + executions.Add(1) + return nil + }) + runtime := &parentChildScenarioRuntime{ + parentID: "parent", + gates: [2]*parentChildReplicationGate{nil, gate}, + delayedTasks: make(map[parentChildReplicationLane]*parentChildReplicationTask), + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + err = delayReplicationAtTaskContainingEvent(initialStandbyCluster, parentWorkflow, enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED).run(ctx, runtime) + cancel() + require.NoError(t, err) + require.Zero(t, executions.Load()) + + runtime.activeClusterIndex = int(initialStandbyCluster) + ctx, cancel = context.WithTimeout(context.Background(), time.Second) + err = applyDelayedReplication(initialStandbyCluster, parentWorkflow).run(ctx, runtime) + cancel() + require.NoError(t, err) + require.NoError(t, receiveParentChildInterceptResult(t, interceptResult)) + require.Equal(t, int32(1), executions.Load()) + require.Contains(t, runtime.trace, " apply delayed task 1 to cluster 1 for parent [WorkflowTaskStarted]") +} + +func TestParentChildHarnessApplyDelayedReplicationRequiresDelayedTask(t *testing.T) { + runtime := &parentChildScenarioRuntime{ + parentID: "parent", + delayedTasks: make(map[parentChildReplicationLane]*parentChildReplicationTask), + } + + err := applyDelayedReplication(initialStandbyCluster, parentWorkflow).run(context.Background(), runtime) + require.ErrorContains(t, err, "no delayed parent replication task to initial standby cluster") +} + +func TestParentChildHarnessVersionedTransitionApplyThrough(t *testing.T) { + gate := newParentChildReplicationGate("namespace", "parent", "child") + defer gate.close() + + var verifyExecutions atomic.Int32 + verifyResult := interceptParentChildTask( + gate, + newParentChildVerifyVersionedTransitionTask("namespace", "parent", "run", 1), + func() error { + verifyExecutions.Add(1) + return nil + }, + ) + await.RequireTrue(t, func() bool { return len(gate.pending) == 1 }, time.Second, time.Millisecond) + + events := []*historypb.HistoryEvent{ + {EventId: 3, EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED}, + {EventId: 4, EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED}, + } + blob, err := serialization.NewSerializer().SerializeEvents(events) + require.NoError(t, err) + var syncExecutions atomic.Int32 + syncResult := interceptParentChildTask( + gate, + newParentChildSyncVersionedTransitionTask("namespace", "parent", "run", 2, blob), + func() error { + syncExecutions.Add(1) + return nil + }, + ) + await.RequireTrue(t, func() bool { return len(gate.pending) == 2 }, time.Second, time.Millisecond) + + runtime := &parentChildScenarioRuntime{ + parentID: "parent", + gates: [2]*parentChildReplicationGate{nil, gate}, + delayedTasks: make(map[parentChildReplicationLane]*parentChildReplicationTask), + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + err = acknowledgeReplicationTaskContainingEventWithoutApplying(initialStandbyCluster, parentWorkflow, enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED).run(ctx, runtime) + cancel() + require.NoError(t, err) + require.NoError(t, receiveParentChildInterceptResult(t, verifyResult)) + require.NoError(t, receiveParentChildInterceptResult(t, syncResult)) + require.Equal(t, int32(1), verifyExecutions.Load()) + require.Zero(t, syncExecutions.Load()) + require.Contains(t, runtime.trace, " apply task 1 to cluster 1 for parent [VerifyVersionedTransition]") + require.Contains(t, runtime.trace, " ack-without-apply task 2 to cluster 1 for parent [WorkflowTaskStarted, WorkflowTaskCompleted]") +} + +func TestParentChildHarnessDecodeReplicationEvents(t *testing.T) { + events := []*historypb.HistoryEvent{ + {EventId: 1, EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED}, + {EventId: 2, EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED}, + } + blob, err := serialization.NewSerializer().SerializeEvents(events) + require.NoError(t, err) + + t.Run("legacy history", func(t *testing.T) { + task := newParentChildHistoryReplicationTask("namespace", "parent", 1) + task.GetHistoryTaskAttributes().Events = blob + actual, err := decodeParentChildReplicationEvents(task) + require.NoError(t, err) + protorequire.ProtoSliceEqual(t, events, actual) + }) + + t.Run("sync versioned transition", func(t *testing.T) { + actual, err := decodeParentChildReplicationEvents( + newParentChildSyncVersionedTransitionTask("namespace", "parent", "run", 1, blob), + ) + require.NoError(t, err) + protorequire.ProtoSliceEqual(t, events, actual) + }) + + t.Run("backfill history", func(t *testing.T) { + actual, err := decodeParentChildReplicationEvents( + newParentChildBackfillHistoryTask("namespace", "parent", "run", 1, blob), + ) + require.NoError(t, err) + protorequire.ProtoSliceEqual(t, events, actual) + }) + + t.Run("verify versioned transition", func(t *testing.T) { + actual, err := decodeParentChildReplicationEvents( + newParentChildVerifyVersionedTransitionTask("namespace", "parent", "run", 1), + ) + require.NoError(t, err) + require.Empty(t, actual) + }) +} + +func newParentChildHistoryReplicationTask( + namespaceID string, + workflowID string, + sourceTaskID int64, +) *replicationspb.ReplicationTask { + return &replicationspb.ReplicationTask{ + SourceTaskId: sourceTaskID, + Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{ + HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{ + NamespaceId: namespaceID, + WorkflowId: workflowID, + }, + }, + } +} + +func newParentChildSyncVersionedTransitionTask( + namespaceID string, + workflowID string, + runID string, + sourceTaskID int64, + eventBatches ...*commonpb.DataBlob, +) *replicationspb.ReplicationTask { + return &replicationspb.ReplicationTask{ + TaskType: enumsspb.REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK, + SourceTaskId: sourceTaskID, + Attributes: &replicationspb.ReplicationTask_SyncVersionedTransitionTaskAttributes{ + SyncVersionedTransitionTaskAttributes: &replicationspb.SyncVersionedTransitionTaskAttributes{ + NamespaceId: namespaceID, + WorkflowId: workflowID, + RunId: runID, + VersionedTransitionArtifact: &replicationspb.VersionedTransitionArtifact{ + EventBatches: eventBatches, + }, + }, + }, + } +} + +func newParentChildVerifyVersionedTransitionTask( + namespaceID string, + workflowID string, + runID string, + sourceTaskID int64, +) *replicationspb.ReplicationTask { + return &replicationspb.ReplicationTask{ + TaskType: enumsspb.REPLICATION_TASK_TYPE_VERIFY_VERSIONED_TRANSITION_TASK, + SourceTaskId: sourceTaskID, + Attributes: &replicationspb.ReplicationTask_VerifyVersionedTransitionTaskAttributes{ + VerifyVersionedTransitionTaskAttributes: &replicationspb.VerifyVersionedTransitionTaskAttributes{ + NamespaceId: namespaceID, + WorkflowId: workflowID, + RunId: runID, + }, + }, + } +} + +func newParentChildBackfillHistoryTask( + namespaceID string, + workflowID string, + runID string, + sourceTaskID int64, + eventBatches ...*commonpb.DataBlob, +) *replicationspb.ReplicationTask { + return &replicationspb.ReplicationTask{ + TaskType: enumsspb.REPLICATION_TASK_TYPE_BACKFILL_HISTORY_TASK, + SourceTaskId: sourceTaskID, + Attributes: &replicationspb.ReplicationTask_BackfillHistoryTaskAttributes{ + BackfillHistoryTaskAttributes: &replicationspb.BackfillHistoryTaskAttributes{ + NamespaceId: namespaceID, + WorkflowId: workflowID, + RunId: runID, + EventBatches: eventBatches, + }, + }, + } +} + +func interceptParentChildTask( + gate *parentChildReplicationGate, + task *replicationspb.ReplicationTask, + execute func() error, +) <-chan error { + result := make(chan error, 1) + go func() { + result <- gate.intercept(task, execute) + }() + return result +} + +func receiveParentChildInterceptResult(t *testing.T, result <-chan error) error { + t.Helper() + select { + case err := <-result: + return err + case <-time.After(time.Second): + t.Fatal("timed out waiting for replication interceptor") + return nil + } +} diff --git a/tests/xdc/parent_child_test.go b/tests/xdc/parent_child_test.go new file mode 100644 index 0000000000..d98c5ade1e --- /dev/null +++ b/tests/xdc/parent_child_test.go @@ -0,0 +1,523 @@ +package xdc + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/suite" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + "go.temporal.io/api/serviceerror" + enumsspb "go.temporal.io/server/api/enums/v1" + "go.temporal.io/server/common" + "go.temporal.io/server/common/metrics" + "go.temporal.io/server/tests/testcore" +) + +func TestParentChildXDCTestSuite(t *testing.T) { + t.Parallel() + suite.Run(t, new(parentChildXDCTestSuite)) +} + +func (s *parentChildXDCTestSuite) SetupSuite() { + s.enableTransitionHistory = true + s.setupSuite(testcore.WithNumHistoryShards(2)) +} + +func (s *parentChildXDCTestSuite) SetupTest() { + s.setupTest() +} + +func (s *parentChildXDCTestSuite) TearDownSuite() { + s.tearDownSuite() +} + +// TestReproOrphanedChildAfterForceFailover covers the cross-shard ordering where the child start +// reaches the passive while parent replication from WorkflowTaskStarted onward, including +// StartChildWorkflowExecutionInitiated, remains delayed, followed by force failover. +// +// | parent on target | child on target | active | outcome +// -----------------------+------------------------------------------+---------------------------------+-----------------+------------------------------ +// parent prefix arrives | WFT scheduled | does not exist | initial active | common parent prefix +// child start arrives | parent update remains delayed | RUNNING, points to old version | initial active | cross-shard partial state +// force failover | incomplete branch becomes current | unchanged | initial standby | target begins active recovery +// retry StartChild | same initiated event ID at a new version | unchanged | initial standby | WORKFLOW_ALREADY_EXISTS +// assert | no ChildWorkflowExecutionStarted | still points to losing version | initial standby | child is orphaned +// +// Event checkpoints select an entire replication task, not an individual event. The delayed parent +// task intentionally remains unapplied through the assertions. +func (s *parentChildXDCTestSuite) TestReproOrphanedChildAfterForceFailover() { + s.runParentChildScenario(parentChildScenario{ + steps: []parentChildScenarioStep{ + // Create the parent and its first workflow task on the initial active cluster. + startParentWorkflow(), + // Give the passive a common parent prefix before introducing the cross-shard gap. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED, + ), + // Complete the parent task with StartChild, creating the child on the source. + completeParentWorkflowTaskWithStartChildCommand(), + // Stop parent replication before its StartChild branch is complete on the passive. + delayReplicationAtTaskContainingEvent( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED, + ), + // Let the child arrive independently, still pointing to the old parent branch. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + childWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, + ), + // Promote the incomplete passive state without draining the delayed parent task. + forceFailoverNamespaceTo(initialStandbyCluster), + // Retry StartChild on the new active branch, where the child ID already exists. + completeParentWorkflowTaskWithStartChildCommand(), + }, + expectations: []parentChildExpectation{ + { + name: "parent StartChild fails because the child already exists", + check: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + events, err := runtime.activeWorkflowHistory(ctx, parentWorkflow) + if err != nil { + return err + } + failedEvent := findHistoryEvent( + events, + enumspb.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_FAILED, + func(event *historypb.HistoryEvent) bool { + attrs := event.GetStartChildWorkflowExecutionFailedEventAttributes() + return attrs.GetWorkflowId() == runtime.childID && + attrs.GetCause() == enumspb.START_CHILD_WORKFLOW_EXECUTION_FAILED_CAUSE_WORKFLOW_ALREADY_EXISTS + }, + ) + if failedEvent == nil { + return fmt.Errorf("parent has no WORKFLOW_ALREADY_EXISTS failure for child %q", runtime.childID) + } + return nil + }, + }, + currentWorkflowHasStatusOnCluster( + initialStandbyCluster, + childWorkflow, + enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, + ), + { + name: "child remains attached to the losing parent branch", + check: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + if runtime.childRunID == "" { + return errors.New("child WorkflowExecutionStarted was not applied") + } + childEvents, err := runtime.activeWorkflowHistory(ctx, childWorkflow) + if err != nil { + return err + } + childStartedEvent := findHistoryEvent( + childEvents, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, + nil, + ) + if childStartedEvent == nil { + return fmt.Errorf("child run %q has no persisted WorkflowExecutionStarted event", runtime.childRunID) + } + childStartedAttrs := childStartedEvent.GetWorkflowExecutionStartedEventAttributes() + if childStartedAttrs == nil { + return errors.New("persisted child start event has no attributes") + } + parentExecution := childStartedAttrs.GetParentWorkflowExecution() + if parentExecution.GetWorkflowId() != runtime.parentID || parentExecution.GetRunId() != runtime.parentRunID { + return fmt.Errorf( + "child parent is %s/%s, want %s/%s", + parentExecution.GetWorkflowId(), + parentExecution.GetRunId(), + runtime.parentID, + runtime.parentRunID, + ) + } + + parentEvents, err := runtime.activeWorkflowHistory(ctx, parentWorkflow) + if err != nil { + return err + } + currentInitiatedEvent := findHistoryEvent( + parentEvents, + enumspb.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED, + func(event *historypb.HistoryEvent) bool { + attrs := event.GetStartChildWorkflowExecutionInitiatedEventAttributes() + return attrs.GetWorkflowId() == runtime.childID && + event.GetEventId() == childStartedAttrs.GetParentInitiatedEventId() + }, + ) + currentChildStartedEvent := findHistoryEvent( + parentEvents, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED, + func(event *historypb.HistoryEvent) bool { + attrs := event.GetChildWorkflowExecutionStartedEventAttributes() + return attrs.GetWorkflowExecution().GetWorkflowId() == runtime.childID && + attrs.GetWorkflowExecution().GetRunId() == runtime.childRunID + }, + ) + if currentChildStartedEvent != nil { + return errors.New("current parent branch owns the applied child") + } + if currentInitiatedEvent == nil { + return fmt.Errorf( + "current parent branch has no StartChild event %d for child %q", + childStartedAttrs.GetParentInitiatedEventId(), + runtime.childID, + ) + } + if currentInitiatedEvent.GetVersion() == childStartedAttrs.GetParentInitiatedEventVersion() { + return fmt.Errorf( + "child and current parent branch have the same initiation version %d", + currentInitiatedEvent.GetVersion(), + ) + } + runtime.tracef( + "orphan confirmed: child parent pointer=(%d, v%d), current parent initiation=(%d, v%d)", + childStartedAttrs.GetParentInitiatedEventId(), + childStartedAttrs.GetParentInitiatedEventVersion(), + currentInitiatedEvent.GetEventId(), + currentInitiatedEvent.GetVersion(), + ) + return nil + }, + }, + }, + }) +} + +// TestStandbyVerifiesMissingChild covers the cross-shard ordering where the parent update +// identifying a started child reaches the passive while the child's WorkflowExecutionStarted +// task remains delayed. +// +// | parent on target | child on target | active | outcome +// ---------------------------+---------------------------------+-----------------+----------------+--------------------------------------------- +// parent prefix arrives | WFT scheduled | does not exist | initial active | common parent prefix +// child start is delayed | unchanged | does not exist | initial active | child start remains unapplied +// parent child-start arrives | ChildWorkflowExecutionStarted | does not exist | initial active | standby verifies the child's first WFT +// verification fails | child-start relationship exists | does not exist | initial active | VerifyFirstWorkflowTaskScheduled: NotFound +// discard window expires | unchanged | does not exist | initial active | standby StartChild task is discarded +// +// Event checkpoints select an entire replication task, not an individual event. The delayed child +// start task intentionally remains unapplied while the standby StartChild task is allowed to expire. +func (s *parentChildXDCTestSuite) TestStandbyVerifiesMissingChild() { + s.runParentChildScenario(parentChildScenario{ + steps: []parentChildScenarioStep{ + // Track source time without the production standby lag so task expiration is observable quickly. + setStandbyClusterDelay(initialStandbyCluster, 0), + // Create the parent and its first workflow task on the initial active cluster. + startParentWorkflow(), + // Establish the parent on the passive before replicating its child relationship. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED, + ), + // Complete the parent task with StartChild, creating the child on the source. + completeParentWorkflowTaskWithStartChildCommand(), + // Keep the child start off the passive so the child is locally missing there. + delayReplicationAtTaskContainingEvent( + initialStandbyCluster, + childWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, + ), + // Apply the parent's child-start record, triggering verification of the missing child. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED, + ), + // Observe the exact missing-child error before shortening the task's discard window. + waitForHistoryVerificationFailureOnCluster( + initialStandbyCluster, + historyClientVerifyFirstWorkflowTask, + &serviceerror.NotFound{}, + ), + // Expire only the pending standby StartChild task; its next attempt should return ErrTaskDiscarded. + setStandbyTaskDiscardDelay( + initialStandbyCluster, + enumsspb.TASK_TYPE_TRANSFER_START_CHILD_EXECUTION, + 0, + ), + }, + expectations: []parentChildExpectation{ + workflowIsMissingOnCluster(initialStandbyCluster, childWorkflow), + taskWasDiscardedOnCluster( + initialStandbyCluster, + metrics.TaskTypeTransferStandbyTaskStartChildExecution, + ), + }, + }) +} + +// TestStandbyVerifiesChildWithoutFirstWorkflowTask covers the cross-shard ordering where the +// child's WorkflowExecutionStarted reaches the passive, its first WorkflowTaskScheduled remains +// delayed, and the parent update identifying the started child then arrives. +// +// | parent on target | child on target | active | outcome +// ---------------------------+---------------------------------+----------------------------------+----------------+---------------------------------------------- +// parent prefix arrives | WFT scheduled | does not exist | initial active | common parent prefix +// child start arrives | unchanged | CREATED/RUNNING, no first WFT | initial active | child execution exists on the target +// child first WFT is delayed | unchanged | unchanged | initial active | WorkflowTaskScheduled remains unapplied +// parent child-start arrives | ChildWorkflowExecutionStarted | unchanged | initial active | standby verifies the child's first WFT +// verification fails | child-start relationship exists | next event ID 2, no scheduled ID | initial active | VerifyFirstWorkflowTaskScheduled: WorkflowNotReady +// discard window expires | unchanged | unchanged | initial active | standby StartChild task is discarded +// +// This scenario uses legacy history replication because it fixes each transaction's event range. +// The gate can therefore apply WorkflowExecutionStarted while keeping the separate +// WorkflowTaskScheduled task delayed through the assertions. +func (s *parentChildXDCTestSuite) TestStandbyVerifiesChildWithoutFirstWorkflowTask() { + s.runParentChildScenario(parentChildScenario{ + steps: []parentChildScenarioStep{ + // Keep child Started and its first WFT in separate event-range replication tasks. + useLegacyHistoryReplication(), + // Track source time without the production standby lag so task expiration is observable quickly. + setStandbyClusterDelay(initialStandbyCluster, 0), + // Create the parent and its first workflow task on the initial active cluster. + startParentWorkflow(), + // Establish the parent on the passive before replicating its child relationship. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED, + ), + // Complete the parent task with StartChild, creating the child on the source. + completeParentWorkflowTaskWithStartChildCommand(), + // Create the child on the passive without advancing through its first WFT. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + childWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, + ), + // Hold the child's first WFT so its passive mutable state remains not ready. + delayReplicationAtTaskContainingEvent( + initialStandbyCluster, + childWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED, + ), + // Apply the parent's child-start record, triggering verification of that partial child. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED, + ), + // Observe the exact not-ready error before shortening the task's discard window. + waitForHistoryVerificationFailureOnCluster( + initialStandbyCluster, + historyClientVerifyFirstWorkflowTask, + &serviceerror.WorkflowNotReady{}, + ), + // Expire only the pending standby StartChild task; its next attempt should return ErrTaskDiscarded. + setStandbyTaskDiscardDelay( + initialStandbyCluster, + enumsspb.TASK_TYPE_TRANSFER_START_CHILD_EXECUTION, + 0, + ), + }, + expectations: []parentChildExpectation{ + { + name: "child exists without its first workflow task on the initial standby cluster", + check: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + mutableState, err := runtime.workflowMutableState(ctx, initialStandbyCluster, childWorkflow) + if err != nil { + return err + } + if state := mutableState.GetExecutionState().GetState(); state != enumsspb.WORKFLOW_EXECUTION_STATE_CREATED { + return fmt.Errorf("child state is %s, want %s", state, enumsspb.WORKFLOW_EXECUTION_STATE_CREATED) + } + if status := mutableState.GetExecutionState().GetStatus(); status != enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING { + return fmt.Errorf("child status is %s, want %s", status, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING) + } + if scheduledEventID := mutableState.GetExecutionInfo().GetWorkflowTaskScheduledEventId(); scheduledEventID != common.EmptyEventID { + return fmt.Errorf("child workflow task scheduled event ID is %d, want %d", scheduledEventID, common.EmptyEventID) + } + if nextEventID := mutableState.GetNextEventId(); nextEventID != common.FirstEventID+1 { + return fmt.Errorf("child next event ID is %d, want %d", nextEventID, common.FirstEventID+1) + } + return nil + }, + }, + taskWasDiscardedOnCluster( + initialStandbyCluster, + metrics.TaskTypeTransferStandbyTaskStartChildExecution, + ), + }, + }) +} + +// TestStandbyDiscardsChildCloseTaskWhenParentCompletionIsMissing covers the cross-shard ordering +// where the child's completion reaches the passive while the parent's corresponding +// ChildWorkflowExecutionCompleted update remains delayed. +// +// | parent on target | child on target | active | outcome +// ---------------------------+-----------------------------------+-----------------+----------------+---------------------------------------------- +// parent-child state arrives | ChildWorkflowExecutionStarted | RUNNING | initial active | both sides initially agree that child is running +// child closes on source | unchanged on target | unchanged | initial active | source parent records child completion +// parent completion delayed | still tracks the running child | unchanged | initial active | parent completion remains unapplied +// child completion arrives | no child-completion event | COMPLETED | initial active | VerifyChildExecutionCompletionRecorded: WorkflowNotReady +// discard window expires | unchanged | COMPLETED | initial active | standby CloseExecution task is discarded +// +// The local parent-verification grace is kept longer than the compressed discard window so this +// scenario exercises the retry-and-discard fallback rather than parent state resend. The delayed +// parent completion task remains unapplied through the assertions. +func (s *parentChildXDCTestSuite) TestStandbyDiscardsChildCloseTaskWhenParentCompletionIsMissing() { + s.runParentChildScenario(parentChildScenario{ + steps: []parentChildScenarioStep{ + // Track source time without the production standby lag so task expiration is observable quickly. + setStandbyClusterDelay(initialStandbyCluster, 0), + // Create the parent and its first workflow task on the initial active cluster. + startParentWorkflow(), + // Establish the parent on the passive before replicating its child relationship. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED, + ), + // Start the child on the source and produce both sides of the relationship. + completeParentWorkflowTaskWithStartChildCommand(), + // Replicate the child so its later completion can run the passive CloseExecution task. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + childWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, + ), + // Let the passive parent record that the child started and is still running. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED, + ), + // Close the child on the source, which also records completion in the source parent. + completeChildWorkflowTask(), + // Wait for and hold the parent completion so the passive still considers the child running. + delayReplicationAtTaskContainingEvent( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED, + ), + // Prevent this scenario from switching to parent state resend before the task is discarded. + setLocalParentVerificationGrace(initialStandbyCluster, time.Hour), + // Apply child completion, triggering verification against the incomplete passive parent. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + childWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED, + ), + // Observe the exact not-ready error before shortening the task's discard window. + waitForHistoryVerificationFailureOnCluster( + initialStandbyCluster, + historyClientVerifyChildCompletion, + &serviceerror.WorkflowNotReady{}, + ), + // Expire only the pending standby CloseExecution task; its next attempt should be discarded. + setStandbyTaskDiscardDelay( + initialStandbyCluster, + enumsspb.TASK_TYPE_TRANSFER_CLOSE_EXECUTION, + 0, + ), + }, + expectations: []parentChildExpectation{ + currentWorkflowHasStatusOnCluster( + initialStandbyCluster, + childWorkflow, + enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, + ), + { + name: "parent completion remains delayed on the initial standby cluster", + check: func(ctx context.Context, runtime *parentChildScenarioRuntime) error { + events, err := runtime.workflowHistoryOnCluster(ctx, initialStandbyCluster, parentWorkflow) + if err != nil { + return err + } + if historyContainsEvent(events, enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED) { + return errors.New("passive parent already contains ChildWorkflowExecutionCompleted") + } + return nil + }, + }, + taskWasDiscardedOnCluster( + initialStandbyCluster, + metrics.TaskTypeTransferStandbyTaskCloseExecution, + ), + }, + }) +} + +// TestStandbyResendsMissingParentWhenChildCloses covers the cross-shard ordering where child +// replication reaches the passive while ordinary parent replication remains delayed. When the +// child's completion arrives, standby verification finds the parent missing and resends its state. +// +// | parent on target | child on target | active | outcome +// -------------------------+-------------------------------------+--------------------------+----------------+----------------------------------------------- +// parent start is delayed | does not exist | does not exist | initial active | ordinary parent replication remains unapplied +// child start arrives | does not exist | RUNNING, points to parent | initial active | cross-shard partial state +// child closes on source | does not exist | unchanged on target | initial active | source parent records child completion +// confirm missing parent | does not exist | unchanged | initial active | establishes the resend precondition +// child completion arrives | restored by parent state sync | COMPLETED | initial active | VerifyChild detects the missing parent +// assert | has ChildWorkflowExecutionCompleted | COMPLETED | initial active | resend observed; parent state restored +// +// The original parent replication task remains delayed. The local verification grace is set to zero +// so VerifyChildExecutionCompletionRecorded requests a parent resend within the test timeout. +func (s *parentChildXDCTestSuite) TestStandbyResendsMissingParentWhenChildCloses() { + s.runParentChildScenario(parentChildScenario{ + steps: []parentChildScenarioStep{ + // Create the parent and its first workflow task on the initial active cluster. + startParentWorkflow(), + // Keep all ordinary parent replication off the passive. + delayReplicationAtTaskContainingEvent( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, + ), + // Start the child on the source while its parent remains absent on the passive. + completeParentWorkflowTaskWithStartChildCommand(), + // Replicate only the child, creating a child whose parent is locally missing. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + childWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, + ), + // Close the child on the source to produce child-completion verification work. + completeChildWorkflowTask(), + // Ensure a state resend will include the parent's child-completion event. + waitForWorkflowEventOnCluster( + initialActiveCluster, + parentWorkflow, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED, + ), + // Confirm the passive still lacks the parent before delivering child completion. + confirmWorkflowIsMissingOnCluster(initialStandbyCluster, parentWorkflow), + // Skip the normal local retry window so the next verification requests a resend. + setLocalParentVerificationGrace(initialStandbyCluster, 0), + // Deliver child completion, causing standby verification to resend the missing parent. + applyReplicationThroughTaskContainingEvent( + initialStandbyCluster, + childWorkflow, + enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED, + ), + }, + expectations: []parentChildExpectation{ + { + name: "parent workflow resend is attempted on the initial standby cluster", + check: func(_ context.Context, runtime *parentChildScenarioRuntime) error { + return runtime.requireCapturedMetric( + initialStandbyCluster, + metrics.ParentWorkflowResendAttempts.Name(), + nil, + ) + }, + }, + workflowHasEventOnCluster( + initialStandbyCluster, + parentWorkflow, + enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED, + ), + }, + }) +}