Resend child workflow async when missing on passive

common/dynamicconfig/constants.go

@@ -2776,8 +2776,8 @@ 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, and for the
-standby CloseExecutionTask's child-to-parent completion verification`,
+		`ReplicationTaskApplyTimeout is the context timeout for replication task apply, and for
+standby parent-child verification resends`,
 	)
 	ParentWorkflowResendMaxInFlight = NewGlobalIntSetting(
 		"history.parentWorkflowResendMaxInFlight",
@@ -2792,6 +2792,21 @@ verifying task retries. This bounds the goroutines this path can create per shar
 		`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.`,
+	)
+	ChildWorkflowResendMaxInFlight = NewGlobalIntSetting(
+		"history.childWorkflowResendMaxInFlight",
+		8,
+		`ChildWorkflowResendMaxInFlight caps how many child workflow resends a shard may run
+concurrently when EnableChildWorkflowResend is on. Attempts beyond the cap are dropped; the
+verifying task retries. This bounds the goroutines this path can create per shard.`,
+	)
+	EnableChildWorkflowResend = NewGlobalBoolSetting(
+		"history.enableChildWorkflowResend",
+		false,
+		`EnableChildWorkflowResend controls whether standby parent-to-child first workflow task
+verification may resend a missing child workflow in the background from the active cluster. When
+disabled, verification remains local-only. StandbyTaskMissingEventsResendDelay plus
+ReplicationTaskApplyTimeout should remain below StandbyTaskMissingEventsDiscardDelay.`,
 	)
 	ReplicationTaskFetcherParallelism = NewGlobalIntSetting(
 		"history.ReplicationTaskFetcherParallelism",

common/metrics/metric_defs.go

@@ -1137,6 +1137,16 @@ var (
 	ParentWorkflowResendLimited = NewCounterDef("parent_workflow_resend_limited")
 	// ParentWorkflowResendLatency measures a resend: cross-cluster state fetch plus local apply.
 	ParentWorkflowResendLatency = NewTimerDef("parent_workflow_resend_latency")
+	// ChildWorkflowResendAttempts counts child resends started by standby first-task verification.
+	ChildWorkflowResendAttempts = NewCounterDef("child_workflow_resend_attempts")
+	// ChildWorkflowResendSkipped counts attempts that found a resend for the same child in flight.
+	ChildWorkflowResendSkipped = NewCounterDef("child_workflow_resend_skipped")
+	// ChildWorkflowResendFailures counts failed resends. Async resends report failure nowhere else.
+	ChildWorkflowResendFailures = NewCounterDef("child_workflow_resend_failures")
+	// ChildWorkflowResendLimited counts resends dropped because the shard was at its in-flight cap.
+	ChildWorkflowResendLimited = NewCounterDef("child_workflow_resend_limited")
+	// ChildWorkflowResendLatency measures a resend: cross-cluster state fetch plus local apply.
+	ChildWorkflowResendLatency = NewTimerDef("child_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")

proto/internal/temporal/server/api/historyservice/v1/request_response.proto

@@ -593,6 +593,7 @@ message VerifyFirstWorkflowTaskScheduledRequest {
   string namespace_id = 1;
   temporal.api.common.v1.WorkflowExecution workflow_execution = 2;
   temporal.server.api.clock.v1.VectorClock clock = 3;
+  bool resend_child = 4;
 }
 
 message VerifyFirstWorkflowTaskScheduledResponse {}

service/history/api/verifychildworkflowcompletionrecorded/api.go

@@ -24,6 +24,7 @@ import (
 	"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/api/workflowresend"
 	"go.temporal.io/server/service/history/consts"
 	historyi "go.temporal.io/server/service/history/interfaces"
 )
@@ -94,7 +95,7 @@ func Invoke(
 	request *historyservice.VerifyChildExecutionCompletionRecordedRequest,
 	workflowConsistencyChecker api.WorkflowConsistencyChecker,
 	shardContext historyi.ShardContext,
-	inFlightResends *InFlightResends,
+	inFlightResends *workflowresend.InFlightResends,
 ) (*historyservice.VerifyChildExecutionCompletionRecordedResponse, error) {
 	namespaceID := namespace.ID(request.GetNamespaceId())
 	if err := api.ValidateNamespaceUUID(namespaceID); err != nil {
@@ -138,7 +139,7 @@ func Invoke(
 	// 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())
+	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",
@@ -162,7 +163,7 @@ func Invoke(
 	resendCtx, cancel := context.WithTimeout(resendCtx, shardContext.GetConfig().ReplicationTaskApplyTimeout())
 	go func() {
 		defer cancel()
-		defer inFlightResends.release(parentKey)
+		defer inFlightResends.Release(parentKey)
 		defer func() {
 			var panicErr error
 			log.CapturePanic(shardContext.GetLogger(), &panicErr)