diff --git a/client/admin/client.go b/client/admin/client.go index e6bb0b01be..2d8180e27e 100644 --- a/client/admin/client.go +++ b/client/admin/client.go @@ -19,6 +19,9 @@ const ( DefaultTimeout = 10 * time.Second * debug.TimeoutMultiplier // DefaultLargeTimeout is the default timeout used to make calls DefaultLargeTimeout = time.Minute * debug.TimeoutMultiplier + // DefaultStateSyncTimeout is a backstop for SyncWorkflowState, which ships a workflow's state + // across clusters. Callers set the real deadline; the smaller one wins. + DefaultStateSyncTimeout = 10 * time.Minute * debug.TimeoutMultiplier ) type clientImpl struct { @@ -44,6 +47,10 @@ func (c *clientImpl) createContext(parent context.Context) (context.Context, con return context.WithTimeout(parent, c.timeout) } +func (c *clientImpl) createContextWithStateSyncTimeout(parent context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(parent, DefaultStateSyncTimeout) +} + func (c *clientImpl) createContextWithLargeTimeout(parent context.Context) (context.Context, context.CancelFunc) { if parent == nil { return context.WithTimeout(context.Background(), c.largeTimeout) diff --git a/client/admin/client_gen.go b/client/admin/client_gen.go index 5a6a1d16fb..cb6466b9d9 100644 --- a/client/admin/client_gen.go +++ b/client/admin/client_gen.go @@ -454,7 +454,7 @@ func (c *clientImpl) SyncWorkflowState( request *adminservice.SyncWorkflowStateRequest, opts ...grpc.CallOption, ) (*adminservice.SyncWorkflowStateResponse, error) { - ctx, cancel := c.createContext(ctx) + ctx, cancel := c.createContextWithStateSyncTimeout(ctx) defer cancel() return c.client.SyncWorkflowState(ctx, request, opts...) } diff --git a/client/history/client.go b/client/history/client.go index 22ab73bcb3..917ef88dd6 100644 --- a/client/history/client.go +++ b/client/history/client.go @@ -32,6 +32,9 @@ var ( const ( // DefaultTimeout is the default timeout used to make calls DefaultTimeout = time.Second * 30 * debug.TimeoutMultiplier + // DefaultStateSyncTimeout is a backstop for SyncWorkflowState, which ships a workflow's state + // across clusters. Callers set the real deadline; the smaller one wins. + DefaultStateSyncTimeout = 10 * time.Minute * debug.TimeoutMultiplier ) type clientImpl struct { @@ -288,6 +291,10 @@ func (c *clientImpl) createContext(parent context.Context) (context.Context, con return context.WithTimeout(parent, c.timeout) } +func (c *clientImpl) createContextWithStateSyncTimeout(parent context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(parent, DefaultStateSyncTimeout) +} + func (c *clientImpl) shardIDFromWorkflowID(namespaceID, workflowID string) int32 { return common.WorkflowIDToHistoryShard(namespaceID, workflowID, c.numberOfShards) } diff --git a/client/history/client_gen.go b/client/history/client_gen.go index a84ae8574f..c594c7f17c 100644 --- a/client/history/client_gen.go +++ b/client/history/client_gen.go @@ -1396,7 +1396,7 @@ func (c *clientImpl) SyncWorkflowState( var response *historyservice.SyncWorkflowStateResponse op := func(ctx context.Context, client historyservice.HistoryServiceClient) error { var err error - ctx, cancel := c.createContext(ctx) + ctx, cancel := c.createContextWithStateSyncTimeout(ctx) defer cancel() response, err = client.SyncWorkflowState(ctx, request, opts...) return err diff --git a/cmd/tools/genrpcwrappers/main.go b/cmd/tools/genrpcwrappers/main.go index c13683b4a2..e374217e44 100644 --- a/cmd/tools/genrpcwrappers/main.go +++ b/cmd/tools/genrpcwrappers/main.go @@ -79,6 +79,12 @@ var ( largeTimeoutContext = map[string]bool{ "client.admin.GetReplicationMessages": true, } + // stateSyncTimeoutContext are the cross-cluster workflow state sync hops, whose callers set a + // deadline that can exceed even the large timeout. DefaultStateSyncTimeout is only a backstop. + stateSyncTimeoutContext = map[string]bool{ + "client.admin.SyncWorkflowState": true, + "client.history.SyncWorkflowState": true, + } longPollRetryPolicy = map[string]string{ "retryableClient.matching.PollWorkflowTaskQueue": "pollPolicy", "retryableClient.matching.PollActivityTaskQueue": "pollPolicy", @@ -439,6 +445,9 @@ func writeTemplatedMethod(w io.Writer, service service, impl string, m reflect.M if largeTimeoutContext[key] { fields["WithLargeTimeout"] = "WithLargeTimeout" } + if stateSyncTimeoutContext[key] { + fields["WithLargeTimeout"] = "WithStateSyncTimeout" + } if impl == "client" { if service.name == "history" { routingOptions := historyRoutingOptions(reqType) @@ -512,7 +521,7 @@ func (c *clientImpl) {{.Method}}( var response {{.ResponseType}} op := func(ctx context.Context, client historyservice.HistoryServiceClient) error { var err error - ctx, cancel := c.createContext(ctx) + ctx, cancel := c.createContext{{or .WithLargeTimeout ""}}(ctx) defer cancel() response, err = client.{{.Method}}(ctx, request, opts...) return err diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go index 32ef0293bf..c78a0f644e 100644 --- a/common/dynamicconfig/constants.go +++ b/common/dynamicconfig/constants.go @@ -2755,7 +2755,22 @@ the number of children greater than or equal to this threshold`, ReplicationTaskApplyTimeout = NewGlobalDurationSetting( "history.ReplicationTaskApplyTimeout", 20*time.Second, - `ReplicationTaskApplyTimeout is the context timeout for replication task apply`, + `ReplicationTaskApplyTimeout is the context timeout for replication task apply, and for the +standby CloseExecutionTask's child-to-parent completion verification`, + ) + ParentWorkflowResendMaxInFlight = NewGlobalIntSetting( + "history.parentWorkflowResendMaxInFlight", + 8, + `ParentWorkflowResendMaxInFlight caps how many parent workflow resends a shard may run +concurrently when EnableAsyncParentWorkflowResend is on. Attempts beyond the cap are dropped; the +verifying task retries. This bounds the goroutines this path can create per shard.`, + ) + EnableAsyncParentWorkflowResend = NewGlobalBoolSetting( + "history.enableAsyncParentWorkflowResend", + false, + `EnableAsyncParentWorkflowResend controls whether the standby child-to-parent completion +verification resends the parent workflow in the background rather than inline, so the verifying task +is not held for the duration of the cross-cluster sync.`, ) ReplicationTaskFetcherParallelism = NewGlobalIntSetting( "history.ReplicationTaskFetcherParallelism", diff --git a/common/metrics/metric_defs.go b/common/metrics/metric_defs.go index 6717397bff..e5aa8e64cd 100644 --- a/common/metrics/metric_defs.go +++ b/common/metrics/metric_defs.go @@ -1117,6 +1117,16 @@ var ( ReplicationTasksFailed = NewCounterDef("replication_tasks_failed") ReplicationTasksBackFill = NewCounterDef("replication_tasks_back_fill") ReplicationTasksBackFillLatency = NewTimerDef("replication_tasks_back_fill_latency") + // ParentWorkflowResendAttempts counts parent resends started by standby completion verification. + ParentWorkflowResendAttempts = NewCounterDef("parent_workflow_resend_attempts") + // ParentWorkflowResendSkipped counts attempts that found a resend for the same parent in flight. + ParentWorkflowResendSkipped = NewCounterDef("parent_workflow_resend_skipped") + // ParentWorkflowResendFailures counts failed resends. Async resends report failure nowhere else. + ParentWorkflowResendFailures = NewCounterDef("parent_workflow_resend_failures") + // ParentWorkflowResendLimited counts resends dropped because the shard was at its in-flight cap. + ParentWorkflowResendLimited = NewCounterDef("parent_workflow_resend_limited") + // ParentWorkflowResendLatency measures a resend: cross-cluster state fetch plus local apply. + ParentWorkflowResendLatency = NewTimerDef("parent_workflow_resend_latency") // ReplicationOrphanedHistoryBranch tracks cases where history branch cleanup was skipped on error // to avoid deleting successfully written history. These orphaned branches will be cleaned up by GC. ReplicationOrphanedHistoryBranch = NewCounterDef("replication_orphaned_history_branch") diff --git a/service/history/api/verifychildworkflowcompletionrecorded/api.go b/service/history/api/verifychildworkflowcompletionrecorded/api.go index 8a600c3160..0e2b73ffc4 100644 --- a/service/history/api/verifychildworkflowcompletionrecorded/api.go +++ b/service/history/api/verifychildworkflowcompletionrecorded/api.go @@ -3,6 +3,7 @@ package verifychildworkflowcompletionrecorded import ( "context" "errors" + "time" commonpb "go.temporal.io/api/common/v1" "go.temporal.io/api/serviceerror" @@ -15,9 +16,13 @@ import ( "go.temporal.io/server/common" "go.temporal.io/server/common/definition" "go.temporal.io/server/common/locks" + "go.temporal.io/server/common/log" + "go.temporal.io/server/common/log/tag" + "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/persistence/transitionhistory" "go.temporal.io/server/common/persistence/versionhistory" + "go.temporal.io/server/common/rpc" "go.temporal.io/server/service/history/api" "go.temporal.io/server/service/history/consts" historyi "go.temporal.io/server/service/history/interfaces" @@ -89,6 +94,7 @@ func Invoke( request *historyservice.VerifyChildExecutionCompletionRecordedRequest, workflowConsistencyChecker api.WorkflowConsistencyChecker, shardContext historyi.ShardContext, + inFlightResends *InFlightResends, ) (*historyservice.VerifyChildExecutionCompletionRecordedResponse, error) { namespaceID := namespace.ID(request.GetNamespaceId()) if err := api.ValidateNamespaceUUID(namespaceID); err != nil { @@ -107,6 +113,99 @@ func Invoke( return nil, errVerify } + metricsHandler := shardContext.GetMetricsHandler() + + // The measured resend, run either inline or in the background. + resend := func(ctx context.Context) (*historyservice.VerifyChildExecutionCompletionRecordedResponse, error) { + metrics.ParentWorkflowResendAttempts.With(metricsHandler).Record(1) + startTime := time.Now().UTC() + resp, err := resendParentAndVerify(ctx, request, workflowConsistencyChecker, shardContext, namespaceID, versionedTransition, versionHistories, errVerify) + metrics.ParentWorkflowResendLatency.With(metricsHandler).Record(time.Since(startTime)) + if err != nil { + recordResendFailure(shardContext, metricsHandler, request, err) + } + return resp, err + } + + if !shardContext.GetConfig().EnableAsyncParentWorkflowResend() { + return resend(ctx) + } + + // The resend can take minutes while the calling standby task's deadline is short, so run it in + // the background and let that task retry until the parent lands. + // + // At most one resend per parent, and at most ParentWorkflowResendMaxInFlight per shard: callers + // retry while an earlier resend runs, so without these a stale parent, or a namespace with many + // of them, would spawn goroutines without bound. + parentKey := definition.NewWorkflowKey(request.NamespaceId, request.ParentExecution.WorkflowId, request.ParentExecution.RunId) + claimed, atCapacity := inFlightResends.tryClaim(parentKey, shardContext.GetConfig().ParentWorkflowResendMaxInFlight()) + if atCapacity { + metrics.ParentWorkflowResendLimited.With(metricsHandler).Record(1) + shardContext.GetLogger().Warn("Dropped parent workflow resend, shard is at its in-flight limit", + tag.WorkflowNamespaceID(request.GetNamespaceId()), + tag.NewStringTag("parent-workflow-id", request.ParentExecution.GetWorkflowId()), + tag.NewStringTag("parent-run-id", request.ParentExecution.GetRunId()), + tag.NewStringTag("child-workflow-id", request.ChildExecution.GetWorkflowId()), + tag.NewStringTag("child-run-id", request.ChildExecution.GetRunId()), + tag.NewInt("max-in-flight", shardContext.GetConfig().ParentWorkflowResendMaxInFlight()), + ) + return nil, errVerify + } + if !claimed { + metrics.ParentWorkflowResendSkipped.With(metricsHandler).Record(1) + return nil, errVerify + } + + // The context is detached from the request, which gRPC cancels when this handler returns, and + // rooted at the shard lifecycle so the work stops with the shard. + resendCtx := rpc.CopyContextValues(shardContext.GetLifecycleContext(), ctx) + resendCtx, cancel := context.WithTimeout(resendCtx, shardContext.GetConfig().ReplicationTaskApplyTimeout()) + go func() { + defer cancel() + defer inFlightResends.release(parentKey) + defer func() { + var panicErr error + log.CapturePanic(shardContext.GetLogger(), &panicErr) + if panicErr != nil { + metrics.ParentWorkflowResendFailures.With(metricsHandler).Record(1) + } + }() + _, _ = resend(resendCtx) + }() + return nil, errVerify +} + +// recordResendFailure reports a failed parent resend. On the asynchronous path no caller receives +// the error, so these are its only signals. +func recordResendFailure( + shardContext historyi.ShardContext, + metricsHandler metrics.Handler, + request *historyservice.VerifyChildExecutionCompletionRecordedRequest, + err error, +) { + metrics.ParentWorkflowResendFailures.With(metricsHandler).Record(1) + shardContext.GetLogger().Error("Failed to resend parent workflow for child completion verification", + tag.WorkflowNamespaceID(request.GetNamespaceId()), + tag.NewStringTag("parent-workflow-id", request.ParentExecution.GetWorkflowId()), + tag.NewStringTag("parent-run-id", request.ParentExecution.GetRunId()), + tag.NewStringTag("child-workflow-id", request.ChildExecution.GetWorkflowId()), + tag.NewStringTag("child-run-id", request.ChildExecution.GetRunId()), + tag.Error(err), + ) +} + +// resendParentAndVerify pulls the parent workflow's state from the source cluster, applies it, and +// re-checks the child's completion. Separate from Invoke so the async path can run it detached. +func resendParentAndVerify( + ctx context.Context, + request *historyservice.VerifyChildExecutionCompletionRecordedRequest, + workflowConsistencyChecker api.WorkflowConsistencyChecker, + shardContext historyi.ShardContext, + namespaceID namespace.ID, + versionedTransition *persistencespb.VersionedTransition, + versionHistories *historyspb.VersionHistories, + errVerify error, +) (*historyservice.VerifyChildExecutionCompletionRecordedResponse, error) { // Resend parent workflow from source cluster clusterMetadata := shardContext.GetClusterMetadata() diff --git a/service/history/api/verifychildworkflowcompletionrecorded/in_flight_resends.go b/service/history/api/verifychildworkflowcompletionrecorded/in_flight_resends.go new file mode 100644 index 0000000000..d97f0c5a00 --- /dev/null +++ b/service/history/api/verifychildworkflowcompletionrecorded/in_flight_resends.go @@ -0,0 +1,44 @@ +package verifychildworkflowcompletionrecorded + +import ( + "sync" + + "go.temporal.io/server/common/definition" +) + +// InFlightResends tracks the parent workflows a shard is currently resending, so concurrent +// verification attempts for the same parent do not each pull its state from the source cluster. A +// resend clones the parent's full mutable state, and callers retry while one is still running. +// +// It also caps how many resends a shard runs at once, bounding the goroutines this path creates. +// +// The zero value is ready to use; hold it by pointer, never copy it. +type InFlightResends struct { + mu sync.Mutex + keys map[definition.WorkflowKey]struct{} +} + +// tryClaim reserves key for the caller. It reports claimed=false when a resend for the same parent +// is already running, or atCapacity=true when the shard already has maxInFlight resends. A caller +// that claims the key must release it when the resend finishes. +func (r *InFlightResends) tryClaim(key definition.WorkflowKey, maxInFlight int) (claimed bool, atCapacity bool) { + r.mu.Lock() + defer r.mu.Unlock() + if _, ok := r.keys[key]; ok { + return false, false + } + if len(r.keys) >= maxInFlight { + return false, true + } + if r.keys == nil { + r.keys = make(map[definition.WorkflowKey]struct{}) + } + r.keys[key] = struct{}{} + return true, false +} + +func (r *InFlightResends) release(key definition.WorkflowKey) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.keys, key) +} diff --git a/service/history/api/verifychildworkflowcompletionrecorded/in_flight_resends_test.go b/service/history/api/verifychildworkflowcompletionrecorded/in_flight_resends_test.go new file mode 100644 index 0000000000..7eea09b59d --- /dev/null +++ b/service/history/api/verifychildworkflowcompletionrecorded/in_flight_resends_test.go @@ -0,0 +1,73 @@ +package verifychildworkflowcompletionrecorded + +import ( + "testing" + + "github.com/stretchr/testify/require" + "go.temporal.io/server/common/definition" +) + +func key(workflowID string) definition.WorkflowKey { + return definition.NewWorkflowKey("ns", workflowID, "run") +} + +func TestInFlightResends_ZeroValueIsUsable(t *testing.T) { + var r InFlightResends // no constructor + claimed, atCapacity := r.tryClaim(key("a"), 8) + require.True(t, claimed) + require.False(t, atCapacity) +} + +func TestInFlightResends_DedupesSameParent(t *testing.T) { + var r InFlightResends + + claimed, atCapacity := r.tryClaim(key("a"), 8) + require.True(t, claimed) + require.False(t, atCapacity) + + // Same parent while the first is still held. + claimed, atCapacity = r.tryClaim(key("a"), 8) + require.False(t, claimed) + require.False(t, atCapacity, "a duplicate is not a capacity problem") + + // A different parent is unaffected. + claimed, _ = r.tryClaim(key("b"), 8) + require.True(t, claimed) + + // Releasing lets the parent be claimed again. + r.release(key("a")) + claimed, _ = r.tryClaim(key("a"), 8) + require.True(t, claimed) +} + +func TestInFlightResends_EnforcesMaxInFlight(t *testing.T) { + var r InFlightResends + + claimed, _ := r.tryClaim(key("a"), 2) + require.True(t, claimed) + claimed, _ = r.tryClaim(key("b"), 2) + require.True(t, claimed) + + // Third distinct parent exceeds the cap. + claimed, atCapacity := r.tryClaim(key("c"), 2) + require.False(t, claimed) + require.True(t, atCapacity) + + // A duplicate of an already-held parent still reports dedup, not capacity. + claimed, atCapacity = r.tryClaim(key("a"), 2) + require.False(t, claimed) + require.False(t, atCapacity) + + // Freeing a slot admits the previously rejected parent. + r.release(key("b")) + claimed, atCapacity = r.tryClaim(key("c"), 2) + require.True(t, claimed) + require.False(t, atCapacity) +} + +func TestInFlightResends_ZeroMaxRejectsEverything(t *testing.T) { + var r InFlightResends + claimed, atCapacity := r.tryClaim(key("a"), 0) + require.False(t, claimed) + require.True(t, atCapacity) +} diff --git a/service/history/configs/config.go b/service/history/configs/config.go index e88323ea09..9fa9d98c01 100644 --- a/service/history/configs/config.go +++ b/service/history/configs/config.go @@ -282,6 +282,8 @@ type Config struct { // The following is used by the new RPC replication stack ReplicationTaskApplyTimeout dynamicconfig.DurationPropertyFn + EnableAsyncParentWorkflowResend dynamicconfig.BoolPropertyFn + ParentWorkflowResendMaxInFlight dynamicconfig.IntPropertyFn ReplicationTaskFetcherParallelism dynamicconfig.IntPropertyFn ReplicationTaskFetcherAggregationInterval dynamicconfig.DurationPropertyFn ReplicationTaskFetcherTimerJitterCoefficient dynamicconfig.FloatPropertyFn @@ -707,6 +709,8 @@ func NewConfig( SendTransientOrSpeculativeWorkflowTaskEvents: dynamicconfig.SendTransientOrSpeculativeWorkflowTaskEvents.Get(dc), ReplicationTaskApplyTimeout: dynamicconfig.ReplicationTaskApplyTimeout.Get(dc), + EnableAsyncParentWorkflowResend: dynamicconfig.EnableAsyncParentWorkflowResend.Get(dc), + ParentWorkflowResendMaxInFlight: dynamicconfig.ParentWorkflowResendMaxInFlight.Get(dc), ReplicationTaskFetcherParallelism: dynamicconfig.ReplicationTaskFetcherParallelism.Get(dc), ReplicationTaskFetcherAggregationInterval: dynamicconfig.ReplicationTaskFetcherAggregationInterval.Get(dc), ReplicationTaskFetcherTimerJitterCoefficient: dynamicconfig.ReplicationTaskFetcherTimerJitterCoefficient.Get(dc), diff --git a/service/history/history_engine.go b/service/history/history_engine.go index 9608177bfa..b36ce959e2 100644 --- a/service/history/history_engine.go +++ b/service/history/history_engine.go @@ -137,6 +137,7 @@ type ( workflowDeleteManager deletemanager.DeleteManager serializer serialization.Serializer workflowConsistencyChecker api.WorkflowConsistencyChecker + parentResends verifychildworkflowcompletionrecorded.InFlightResends chasmEngine chasm.Engine versionChecker headers.VersionChecker versionCache worker_versioning.VersionMembershipAndReactivationStatusCache @@ -742,7 +743,7 @@ func (e *historyEngineImpl) VerifyChildExecutionCompletionRecorded( ctx context.Context, req *historyservice.VerifyChildExecutionCompletionRecordedRequest, ) (*historyservice.VerifyChildExecutionCompletionRecordedResponse, error) { - return verifychildworkflowcompletionrecorded.Invoke(ctx, req, e.workflowConsistencyChecker, e.shardContext) + return verifychildworkflowcompletionrecorded.Invoke(ctx, req, e.workflowConsistencyChecker, e.shardContext, &e.parentResends) } func (e *historyEngineImpl) ReplicateEventsV2( diff --git a/service/history/history_engine2_test.go b/service/history/history_engine2_test.go index 0fbcd0b750..c9c3689409 100644 --- a/service/history/history_engine2_test.go +++ b/service/history/history_engine2_test.go @@ -66,6 +66,7 @@ import ( wcache "go.temporal.io/server/service/history/workflow/cache" "go.temporal.io/server/service/worker/workerdeployment" "go.uber.org/mock/gomock" + "google.golang.org/grpc" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -2578,6 +2579,8 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_WorkflowNotExi } func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParent() { + // Inline resend: the RPC pulls and re-verifies before returning. + s.config.EnableAsyncParentWorkflowResend = func() bool { return false } request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{ NamespaceId: tests.ParentNamespaceID.String(), @@ -2665,6 +2668,112 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParent() s.NoError(err) } +// Async resend: the RPC returns the verification error immediately and the pull runs in the +// background. +func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentAsync() { + s.config.EnableAsyncParentWorkflowResend = func() bool { return true } + + request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{ + NamespaceId: tests.ParentNamespaceID.String(), + ParentExecution: &commonpb.WorkflowExecution{ + WorkflowId: tests.WorkflowID, + RunId: tests.RunID, + }, + ChildExecution: &commonpb.WorkflowExecution{ + WorkflowId: "child workflowId", + RunId: "child runId", + }, + ParentInitiatedId: 123, + ParentInitiatedVersion: 100, + ResendParent: true, + } + + // Parent is absent locally, so verification fails and a resend is eligible. + s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, &serviceerror.NotFound{}).AnyTimes() + + mockClusterMetadata := cluster.NewMockMetadata(s.controller) + mockClusterMetadata.EXPECT().GetClusterID().Return(tests.Version).AnyTimes() + mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestAlternativeClusterName).AnyTimes() + mockClusterMetadata.EXPECT().GetAllClusterInfo().Return(cluster.TestAllClusterInfo).AnyTimes() + mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(true, tests.Version).Return(cluster.TestCurrentClusterName).AnyTimes() + s.mockShard.SetClusterMetadata(mockClusterMetadata) + + // Signal when the background resend lands, so assertions do not race the goroutine. + syncCalled := make(chan struct{}) + s.mockShard.Resource.RemoteAdminClient.EXPECT().SyncWorkflowState(gomock.Any(), gomock.Any()). + DoAndReturn(func(context.Context, *adminservice.SyncWorkflowStateRequest, ...grpc.CallOption) (*adminservice.SyncWorkflowStateResponse, error) { + close(syncCalled) + return nil, serviceerror.NewUnavailable("source cluster unavailable") + }).Times(1) + + // The RPC itself returns the verification error without waiting for the resend. + _, err := s.historyEngine.VerifyChildExecutionCompletionRecorded(metrics.AddMetricsContext(context.Background()), request) + var notFound *serviceerror.NotFound + s.ErrorAs(err, ¬Found) + + select { + case <-syncCalled: + case <-time.After(10 * time.Second): + s.Fail("background resend did not call SyncWorkflowState") + } +} + +// TestVerifyChildExecutionCompletionRecorded_ResendParentDeduped asserts that a second attempt for +// the same parent does not start a concurrent resend while the first is still running. +func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentDeduped() { + s.config.EnableAsyncParentWorkflowResend = func() bool { return true } + + request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{ + NamespaceId: tests.ParentNamespaceID.String(), + ParentExecution: &commonpb.WorkflowExecution{ + WorkflowId: tests.WorkflowID, + RunId: tests.RunID, + }, + ChildExecution: &commonpb.WorkflowExecution{ + WorkflowId: "child workflowId", + RunId: "child runId", + }, + ParentInitiatedId: 123, + ParentInitiatedVersion: 100, + ResendParent: true, + } + + s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, &serviceerror.NotFound{}).AnyTimes() + + mockClusterMetadata := cluster.NewMockMetadata(s.controller) + mockClusterMetadata.EXPECT().GetClusterID().Return(tests.Version).AnyTimes() + mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestAlternativeClusterName).AnyTimes() + mockClusterMetadata.EXPECT().GetAllClusterInfo().Return(cluster.TestAllClusterInfo).AnyTimes() + mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(true, tests.Version).Return(cluster.TestCurrentClusterName).AnyTimes() + s.mockShard.SetClusterMetadata(mockClusterMetadata) + + // Times(1): only one resend may reach the source. It blocks so the second attempt overlaps. + entered := make(chan struct{}) + release := make(chan struct{}) + s.mockShard.Resource.RemoteAdminClient.EXPECT().SyncWorkflowState(gomock.Any(), gomock.Any()). + DoAndReturn(func(context.Context, *adminservice.SyncWorkflowStateRequest, ...grpc.CallOption) (*adminservice.SyncWorkflowStateResponse, error) { + close(entered) + <-release + return nil, serviceerror.NewUnavailable("source cluster unavailable") + }).Times(1) + + ctx := metrics.AddMetricsContext(context.Background()) + _, err := s.historyEngine.VerifyChildExecutionCompletionRecorded(ctx, request) + s.Error(err) + + select { + case <-entered: + case <-time.After(10 * time.Second): + s.Fail("first resend did not reach the source cluster") + } + + // Second attempt while the first is in flight: must not start another resend. + _, err = s.historyEngine.VerifyChildExecutionCompletionRecorded(ctx, request) + s.Error(err) + + close(release) +} + func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_WorkflowClosed() { request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{