mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
XDC coverage for parent-child replication edge cases (#11690)
## What changed? - Added a reusable parent-child XDC test harness for controlling replication task application, delay, ordering, and omission around namespace failover. - Added five functional scenarios covering orphaned children, missing or incomplete children, task discard, missing parent completion, and parent resend recovery. - Added unit tests for the replication gate and legacy/transition-history task decoding. Configured two history shards per cluster and placed parent and child workflows on different shards. - Updated XDC synchronization checks to support multiple history shards. ## Why? Parent-child replication failures depend on rare cross-shard ordering and failover timing, making them difficult to reproduce reliably. These tests deterministically construct the relevant partial states while still exercising real Temporal services, persistence, replication, verification RPCs, task retry/discard behavior, and namespace failover. The harness also makes future scenarios easier to add and review. ## How did you test it? - [x] built - [x] run locally and tested manually - [ ] covered by existing tests - [x] added new unit test(s) - [x] added new functional test(s)
This commit is contained in:
Jiechen Zhong
@@ -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))
|
||||
}
|
||||
|
||||
|
||||
1262
tests/xdc/parent_child_harness.go
Normal file
1262
tests/xdc/parent_child_harness.go
Normal file
File diff suppressed because it is too large
Load Diff
398
tests/xdc/parent_child_harness_test.go
Normal file
398
tests/xdc/parent_child_harness_test.go
Normal file
@@ -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
|
||||
}
|
||||
}
|
||||
523
tests/xdc/parent_child_test.go
Normal file
523
tests/xdc/parent_child_test.go
Normal file
@@ -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,
|
||||
),
|
||||
},
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user