Resend parent workflow asynchronously during standby child completion verification (#11424)

## Problem

A standby child workflow's `CloseExecutionTask` verifies its parent
recorded the completion, and past
`MaxLocalParentWorkflowVerificationDuration` also resends the parent
from the active cluster. That resend is a cross-cluster state sync plus
a possibly paginated history backfill — minutes of work — but the whole
call was bounded by the standby task's hard-coded **3s** `taskTimeout`.
Measured at the active cluster's history shard, the deadline arriving
there was `2.999s`. So the resend never completed.

## Change

- **Run the resend in the background**, bounded by
`ReplicationTaskApplyTimeout` — the setting that already bounds this
same work on the replication stream. The verify RPC returns immediately
and the standby task retries until the parent lands, so it never holds a
transfer-queue worker for the sync.
- The background context is **detached from the request** (gRPC cancels
that when the handler returns) and **rooted at the shard lifecycle**, so
the work stops with the shard.
- **One in-flight resend per parent**, tracked in a shard-level map, and
at most `history.parentWorkflowResendMaxInFlight` (8) concurrent resends
per shard. Callers retry while an earlier resend runs; without this the
test measured 5 full state fetches where 1 suffices. The cap bounds the
goroutines this path can create.
- **Lifted two client ceilings** so the deadline can actually propagate
— `admin.SyncWorkflowState` (was 10s) and `history.SyncWorkflowState`
(was 30s) now share a `DefaultStateSyncTimeout` backstop. This also
fixes the same 10s cap on the replication stream's
`ExecutableTaskImpl.SyncState`, where production's 5m setting was never
reachable either.
- Metrics:
`parent_workflow_resend_{attempts,skipped,limited,failures,latency}`.
Async failures reach no caller, so `_failures` is the alert signal;
`_limited` means the shard is shedding resends. The background goroutine
recovers panics, which would otherwise take down the process.

Also fixes the history-client codegen template, which hardcoded
`createContext` and silently ignored the timeout-tier field.

## Rollout

`history.enableAsyncParentWorkflowResend`, **default false**. Disabled =
the previous inline behavior, bounded by the caller's task deadline. Opt
in per cell.

## Testing

Unit tests cover the inline, async, and per-parent-dedup paths.

An xdc test (added in abae9cec, removed in b6eb6631) withholds the
parent's replication tasks so the child *must* pull it, asserts the
parent is absent from the standby, then stalls the active cluster's
`SyncWorkflowState` for 4 minutes:

```
--- PASS: TestChildPullsParentWhenParentReplicationIsWithheld (286.03s)
incoming-ctx-remaining: 4m59.999859334s    (2.999s before this change)
sync-state-calls:       1                  (5 without the per-parent guard)
dropped-parent-tasks:   9
```

During the stall, 4 verify RPCs reached the standby parent shard (t+0,
+50s, +101s, +169s) and exactly 1 `SyncWorkflowState` reached the active
cluster: the task retried and the guard turned the retries away.

4 minutes exceeds every deadline that previously bounded this path (3s /
10s / 30s) with ~1m headroom against the 5m setting, so the setting is
demonstrably what governs.

To reproduce: `git revert b6eb6631`, then
`go test -tags test_dep ./tests/xdc/ -run
TestVerifyChildCompletionParentResendSuite -timeout 30m`

## Known gaps

- Concurrency across *distinct* parents is unbounded (ordinary fan-out,
not amplification).
- When the parent is deleted on the source, the async path can't report
that back, so the child retries to the 15m discard instead of finishing
immediately. The `workflowNotFoundCache` TODO already in this file would
address it.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

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)

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...)
 }

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)
 }

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

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

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",

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")

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()

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)
+}

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)
+}

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),

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(

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, &notFound)
+
+	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{