From 534045279e15856b4274058ff4758dd5591417b1 Mon Sep 17 00:00:00 2001 From: Jiechen Zhong Date: Tue, 25 Aug 2026 16:00:13 -0700 Subject: [PATCH] Add wide events for parent-child workflow lifecycle debugging (#11707) ## What changed? - Added `parent_child_lifecycle` wide events at critical parent-child failure and recovery points. - Recorded parent/child execution identities, initiated event ID/version, local task metadata, errors, and locally observed workflow state such as `Zombie`. - Added visibility into standby verification and parent resend scheduling, execution, deduplication, and completion. - Reused the existing `history.emitReplicationLifecycleEvents` flag. - Added and extended unit tests for event payloads and emission paths. ## Why? Parent-child failures across shards and clusters are difficult to diagnose from existing metrics alone. These events help distinguish missing workflows, missing first workflow tasks, completion verification failures, workflow ID conflicts, orphaned children, and parent resend progress. They can also be correlated with replication lifecycle and task-discard events. ## How did you test it? - [x] built - [ ] run locally and tested manually - [x] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s) ### output from local e2e test
Wide Events for Parent resend scenario ``` [ { "attempt": 1, "details": { "child_workflow_state": "Completed", "event_type": "parent_child_lifecycle", "local_task_id": 1048587, "message": "Standby child completion verification started with parent resend requested", "operation": "standby_verification", "outcome": "started", "parent_initiated_version": 1, "parent_namespace_id": "691a916b-7f7a-4284-93af-782495421958", "phase": "verify_child_completion", "version": 1 }, "event_name": "replication_lifecycle", "namespace": "test-namespace-974d10a5-e0ef-451a-bac1-0834ec45f658", "namespace_id": "691a916b-7f7a-4284-93af-782495421958", "parent_initiated_id": 5, "parent_run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "parent_workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee", "phase": "executing", "run_id": "01a034ee-475b-7e96-9e3f-4239bd23f79a", "shard": 2, "task_type": "TransferCloseExecution", "workflow_id": "child-211e426f-e194-4492-8534-32d30fefb827" }, { "attempt": 0, "details": { "child_run_id": "01a034ee-475b-7e96-9e3f-4239bd23f79a", "child_workflow_id": "child-211e426f-e194-4492-8534-32d30fefb827", "event_type": "parent_child_lifecycle", "initial_error_type": "serviceerror.NotFound", "local_cluster": "standby_annmr", "message": "Parent workflow resend checkpoint", "operation": "standby_verification_sync_state", "outcome": "started", "parent_initiated_id": 5, "parent_initiated_version": 1, "parent_namespace_id": "691a916b-7f7a-4284-93af-782495421958", "parent_run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "parent_workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee", "parent_workflow_state": "", "phase": "parent_resend", "stage": "sync_workflow_state" }, "event_name": "replication_lifecycle", "namespace": "test-namespace-974d10a5-e0ef-451a-bac1-0834ec45f658", "namespace_id": "691a916b-7f7a-4284-93af-782495421958", "phase": "executing", "run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "shard": 1, "source_cluster": "active_annmr", "task_type": "sync_workflow_state", "workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee" }, { "details": { "child_run_id": "01a034ee-475b-7e96-9e3f-4239bd23f79a", "child_workflow_id": "child-211e426f-e194-4492-8534-32d30fefb827", "event_type": "parent_child_lifecycle", "initial_error_type": "serviceerror.NotFound", "local_cluster": "standby_annmr", "message": "Parent workflow resend checkpoint", "operation": "standby_verification_sync_state", "outcome": "succeeded", "parent_initiated_id": 5, "parent_initiated_version": 1, "parent_namespace_id": "691a916b-7f7a-4284-93af-782495421958", "parent_run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "parent_workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee", "parent_workflow_state": "Running", "phase": "parent_resend" }, "event_name": "replication_lifecycle", "namespace": "test-namespace-974d10a5-e0ef-451a-bac1-0834ec45f658", "namespace_id": "691a916b-7f7a-4284-93af-782495421958", "outcome": "verified", "phase": "applied", "run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "shard": 1, "source_cluster": "active_annmr", "task_type": "sync_workflow_state", "workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee" }, { "details": { "attempt": 1, "child_workflow_state": "Completed", "event_type": "parent_child_lifecycle", "local_task_id": 1048587, "message": "Standby child completion verification completed", "operation": "standby_verification", "outcome": "verified", "parent_initiated_version": 1, "parent_namespace_id": "691a916b-7f7a-4284-93af-782495421958", "phase": "verify_child_completion", "version": 1 }, "event_name": "replication_lifecycle", "namespace": "test-namespace-974d10a5-e0ef-451a-bac1-0834ec45f658", "namespace_id": "691a916b-7f7a-4284-93af-782495421958", "outcome": "verified", "parent_initiated_id": 5, "parent_run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "parent_workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee", "phase": "applied", "run_id": "01a034ee-475b-7e96-9e3f-4239bd23f79a", "shard": 2, "task_type": "TransferCloseExecution", "workflow_id": "child-211e426f-e194-4492-8534-32d30fefb827" } ] ```
## Potential risks Enabling replication lifecycle events will produce additional wide-event volume. --- common/dynamicconfig/constants.go | 2 +- common/wideevents/replication_events.go | 21 ++ .../api.go | 175 +++++++++++++-- service/history/history_engine2_test.go | 158 +++++++++++++- service/history/parent_child_events.go | 205 ++++++++++++++++++ service/history/parent_child_events_test.go | 80 +++++++ service/history/shard/context_testutil.go | 16 +- .../transfer_queue_standby_task_executor.go | 31 ++- ...ansfer_queue_standby_task_executor_test.go | 94 +++++++- 9 files changed, 748 insertions(+), 34 deletions(-) create mode 100644 service/history/parent_child_events.go create mode 100644 service/history/parent_child_events_test.go diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go index c253135b77..26ba8ec802 100644 --- a/common/dynamicconfig/constants.go +++ b/common/dynamicconfig/constants.go @@ -1739,7 +1739,7 @@ branch token is not the execution's current one, but still serves the read.`, EmitReplicationLifecycleEvents = NewGlobalBoolSetting( "history.emitReplicationLifecycleEvents", false, - `EmitReplicationLifecycleEvents controls whether the history service emits ReplicationLifecycle wide events (sent/executing/applied/skipped/error phases). Cluster-level; default off.`, + `EmitReplicationLifecycleEvents controls whether the history service emits ReplicationLifecycle wide events, including parent-child diagnostics. Cluster-level; default off.`, ) EnableCloseInboundReplicationStreamOnShutdown = NewGlobalBoolSetting( "history.enableCloseInboundReplicationStreamOnShutdown", diff --git a/common/wideevents/replication_events.go b/common/wideevents/replication_events.go index aff0342edf..3621db1641 100644 --- a/common/wideevents/replication_events.go +++ b/common/wideevents/replication_events.go @@ -104,6 +104,27 @@ const ( ReplOperationDLQWrite = "dlq_write" ) +// Parent-child diagnostics reuse the ReplicationLifecycle event and store this vocabulary in +// details. Keeping the values here gives producers a shared contract without adding an event table. +const ( + ParentChildLifecycleEventType = "parent_child_lifecycle" + + ParentChildPhaseVerifyChildCompletion = "verify_child_completion" + ParentChildPhaseParentResend = "parent_resend" + + ParentChildOutcomeNotFound = "not_found" + ParentChildOutcomeCompletionMissing = "completion_missing" + ParentChildOutcomeScheduled = "scheduled" + ParentChildOutcomeStarted = "started" + ParentChildOutcomeVerified = "verified" + ParentChildOutcomeIgnored = "ignored" + ParentChildOutcomeSucceeded = "succeeded" + ParentChildOutcomeSourceNotFound = "source_not_found" + ParentChildOutcomeDeduplicated = "deduplicated" + ParentChildOutcomeLimited = "limited" + ParentChildOutcomeFailed = "failed" +) + // EmitReplicationError normalizes the common error envelope and emits it into the existing // replication_lifecycle event, not a separate event table. func EmitReplicationError( diff --git a/service/history/api/verifychildworkflowcompletionrecorded/api.go b/service/history/api/verifychildworkflowcompletionrecorded/api.go index 65f01388e5..5a1b59da0c 100644 --- a/service/history/api/verifychildworkflowcompletionrecorded/api.go +++ b/service/history/api/verifychildworkflowcompletionrecorded/api.go @@ -3,6 +3,7 @@ package verifychildworkflowcompletionrecorded import ( "context" "errors" + "maps" "time" commonpb "go.temporal.io/api/common/v1" @@ -23,6 +24,8 @@ import ( "go.temporal.io/server/common/persistence/transitionhistory" "go.temporal.io/server/common/persistence/versionhistory" "go.temporal.io/server/common/rpc" + "go.temporal.io/server/common/util" + "go.temporal.io/server/common/wideevents" "go.temporal.io/server/service/history/api" "go.temporal.io/server/service/history/consts" historyi "go.temporal.io/server/service/history/interfaces" @@ -33,7 +36,10 @@ func verifyChildExecution( workflowConsistencyChecker api.WorkflowConsistencyChecker, request *historyservice.VerifyChildExecutionCompletionRecordedRequest, ) (versionedTransition *persistencespb.VersionedTransition, - versionHistories *historyspb.VersionHistories, retError error) { + versionHistories *historyspb.VersionHistories, + parentWorkflowState string, + retError error, +) { workflowLease, err := workflowConsistencyChecker.GetWorkflowLease( ctx, request.Clock, @@ -49,28 +55,29 @@ func verifyChildExecution( locks.PriorityLow, ) if err != nil { - return nil, nil, err + return nil, nil, "", err } defer func() { workflowLease.GetReleaseFn()(retError) }() mutableState := workflowLease.GetMutableState() + parentWorkflowState = mutableState.GetExecutionState().GetState().String() if !mutableState.IsWorkflowExecutionRunning() && mutableState.GetExecutionState().State != enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE { // parent has already completed and can't be blocked after failover. - return nil, nil, nil + return nil, nil, parentWorkflowState, nil } onCurrentBranch, err := api.IsHistoryEventOnCurrentBranch(mutableState, request.ParentInitiatedId, request.ParentInitiatedVersion) if err != nil { // initiated event not found on any branch - return nil, nil, consts.ErrWorkflowNotReady + return nil, nil, parentWorkflowState, consts.ErrWorkflowNotReady } if !onCurrentBranch { // due to conflict resolution, the initiated event may on a different branch of the workflow. // we don't have to do anything and can simply return not found error. Standby logic // after seeing this error will give up verification. - return nil, nil, consts.ErrChildExecutionNotFound + return nil, nil, parentWorkflowState, consts.ErrChildExecutionNotFound } ci, isRunning := mutableState.GetChildExecutionInfo(request.ParentInitiatedId) @@ -78,15 +85,15 @@ func verifyChildExecution( if ci.StartedEventId != common.EmptyEventID && ci.GetStartedWorkflowId() != request.ChildExecution.GetWorkflowId() { // this can happen since we may not have the initiated version - return nil, nil, consts.ErrChildExecutionNotFound + return nil, nil, parentWorkflowState, consts.ErrChildExecutionNotFound } - return nil, nil, consts.ErrWorkflowNotReady + return nil, nil, parentWorkflowState, consts.ErrWorkflowNotReady } versionedTransition = transitionhistory.CopyVersionedTransition(transitionhistory.LastVersionedTransition(mutableState.GetExecutionInfo().TransitionHistory)) versionHistories = versionhistory.CopyVersionHistories(mutableState.GetExecutionInfo().VersionHistories) - return versionedTransition, versionHistories, nil + return versionedTransition, versionHistories, parentWorkflowState, nil } func Invoke( @@ -102,7 +109,7 @@ func Invoke( } resendParent := false - versionedTransition, versionHistories, errVerify := verifyChildExecution(ctx, workflowConsistencyChecker, request) + versionedTransition, versionHistories, parentWorkflowState, errVerify := verifyChildExecution(ctx, workflowConsistencyChecker, request) switch errVerify.(type) { case nil: return &historyservice.VerifyChildExecutionCompletionRecordedResponse{}, nil @@ -114,12 +121,14 @@ func Invoke( } metricsHandler := shardContext.GetMetricsHandler() + emitLifecycle := shardContext.GetConfig().EmitReplicationLifecycleEvents() + asyncResend := shardContext.GetConfig().EnableAsyncParentWorkflowResend() // 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) + resp, err := resendParentAndVerify(ctx, request, workflowConsistencyChecker, shardContext, namespaceID, versionedTransition, versionHistories, errVerify, parentWorkflowState, emitLifecycle) metrics.ParentWorkflowResendLatency.With(metricsHandler).Record(time.Since(startTime)) if err != nil { recordResendFailure(shardContext, metricsHandler, request, err) @@ -127,7 +136,7 @@ func Invoke( return resp, err } - if !shardContext.GetConfig().EnableAsyncParentWorkflowResend() { + if !asyncResend { return resend(ctx) } @@ -138,9 +147,15 @@ 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()) + maxInFlight := shardContext.GetConfig().ParentWorkflowResendMaxInFlight() + claimed, atCapacity := inFlightResends.tryClaim(parentKey, maxInFlight) if atCapacity { metrics.ParentWorkflowResendLimited.With(metricsHandler).Record(1) + if emitLifecycle { + details := parentResendEventDetails(errVerify) + details["max_in_flight"] = maxInFlight + emitParentResendLifecycleEvent(shardContext, request, parentWorkflowState, wideevents.ParentChildOutcomeLimited, nil, details) + } 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()), @@ -153,8 +168,28 @@ func Invoke( } if !claimed { metrics.ParentWorkflowResendSkipped.With(metricsHandler).Record(1) + if emitLifecycle { + emitParentResendLifecycleEvent( + shardContext, + request, + parentWorkflowState, + wideevents.ParentChildOutcomeDeduplicated, + nil, + parentResendEventDetails(errVerify), + ) + } return nil, errVerify } + if emitLifecycle { + emitParentResendLifecycleEvent( + shardContext, + request, + parentWorkflowState, + wideevents.ParentChildOutcomeScheduled, + nil, + parentResendEventDetails(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. @@ -205,26 +240,47 @@ func resendParentAndVerify( versionedTransition *persistencespb.VersionedTransition, versionHistories *historyspb.VersionHistories, errVerify error, + parentWorkflowState string, + emitLifecycle bool, ) (*historyservice.VerifyChildExecutionCompletionRecordedResponse, error) { // Resend parent workflow from source cluster clusterMetadata := shardContext.GetClusterMetadata() targetClusterInfo := clusterMetadata.GetAllClusterInfo()[clusterMetadata.GetCurrentClusterName()] + activeClusterName := "" + emitResult := func(outcome string, eventErr error, stage string) { + if !emitLifecycle { + return + } + details := parentResendEventDetails(errVerify) + if activeClusterName != "" { + details["source_cluster"] = activeClusterName + } + if stage != "" { + details["stage"] = stage + } + emitParentResendLifecycleEvent(shardContext, request, parentWorkflowState, outcome, eventErr, details) + } namespaceEntry, err := shardContext.GetNamespaceRegistry().GetNamespaceByID(namespaceID) if err != nil { + emitResult(wideevents.ParentChildOutcomeFailed, err, "resolve_namespace") return nil, err } - activeClusterName := namespaceEntry.ActiveClusterName(namespace.RoutingKey{ID: request.ParentExecution.WorkflowId}) + activeClusterName = namespaceEntry.ActiveClusterName(namespace.RoutingKey{ID: request.ParentExecution.WorkflowId}) if activeClusterName == clusterMetadata.GetCurrentClusterName() { - return nil, errors.New("namespace becomes active when processing task as standby") + err = errors.New("namespace becomes active when processing task as standby") + emitResult(wideevents.ParentChildOutcomeFailed, err, "resolve_source_cluster") + return nil, err } remoteAdminClient, err := shardContext.GetRemoteAdminClient(activeClusterName) if err != nil { + emitResult(wideevents.ParentChildOutcomeFailed, err, "resolve_remote_client") return nil, err } + emitResult(wideevents.ParentChildOutcomeStarted, nil, "sync_workflow_state") resp, err := remoteAdminClient.SyncWorkflowState(ctx, &adminservice.SyncWorkflowStateRequest{ NamespaceId: request.NamespaceId, @@ -243,30 +299,119 @@ func resendParentAndVerify( // parent workflow is not found on source cluster, // we can return empty response to indicate that verification is done // TODO: add parent workflow to workflowNotFoundCache + emitResult(wideevents.ParentChildOutcomeSourceNotFound, err, "sync_workflow_state") return &historyservice.VerifyChildExecutionCompletionRecordedResponse{}, nil } if _, ok := errors.AsType[*serviceerror.FailedPrecondition](err); ok { // Unable to perform sync state. Transition history maybe disabled in source cluster. + emitResult(wideevents.ParentChildOutcomeFailed, err, "sync_workflow_state") return nil, errVerify } + emitResult(wideevents.ParentChildOutcomeFailed, err, "sync_workflow_state") return nil, err } engine, err := shardContext.GetEngine(ctx) if err != nil { + emitResult(wideevents.ParentChildOutcomeFailed, err, "get_engine") return nil, err } err = engine.ReplicateVersionedTransition(ctx, chasm.WorkflowArchetypeID, resp.VersionedTransitionArtifact, activeClusterName) if err != nil { if !errors.Is(err, consts.ErrDuplicate) { + emitResult(wideevents.ParentChildOutcomeFailed, err, "replicate_versioned_transition") return nil, err } } // Verify child execution again after resending parent workflow - _, _, err = verifyChildExecution(ctx, workflowConsistencyChecker, request) + _, _, observedParentWorkflowState, err := verifyChildExecution(ctx, workflowConsistencyChecker, request) + if observedParentWorkflowState != "" { + parentWorkflowState = observedParentWorkflowState + } if err != nil { + emitResult(wideevents.ParentChildOutcomeFailed, err, "verify_after_resend") return nil, err } + emitResult(wideevents.ParentChildOutcomeSucceeded, nil, "") return &historyservice.VerifyChildExecutionCompletionRecordedResponse{}, nil } + +func parentResendEventDetails(initialError error) map[string]any { + return map[string]any{ + "initial_error_type": util.ErrorType(initialError), + } +} + +func emitParentResendLifecycleEvent( + shardContext historyi.ShardContext, + request *historyservice.VerifyChildExecutionCompletionRecordedRequest, + parentWorkflowState string, + outcome string, + err error, + details map[string]any, +) { + eventDetails := make(map[string]any, len(details)+11) + maps.Copy(eventDetails, details) + eventDetails["event_type"] = wideevents.ParentChildLifecycleEventType + eventDetails["phase"] = wideevents.ParentChildPhaseParentResend + eventDetails["outcome"] = outcome + eventDetails["local_cluster"] = shardContext.GetClusterMetadata().GetCurrentClusterName() + eventDetails["parent_namespace_id"] = request.GetNamespaceId() + eventDetails["parent_workflow_id"] = request.GetParentExecution().GetWorkflowId() + eventDetails["parent_run_id"] = request.GetParentExecution().GetRunId() + eventDetails["parent_workflow_state"] = parentWorkflowState + eventDetails["child_workflow_id"] = request.GetChildExecution().GetWorkflowId() + eventDetails["child_run_id"] = request.GetChildExecution().GetRunId() + eventDetails["parent_initiated_id"] = request.GetParentInitiatedId() + eventDetails["parent_initiated_version"] = request.GetParentInitiatedVersion() + + namespaceName := "" + if entry, namespaceErr := shardContext.GetNamespaceRegistry().GetNamespaceByID(namespace.ID(request.GetNamespaceId())); namespaceErr == nil { + namespaceName = entry.Name().String() + } + sourceCluster, ok := eventDetails["source_cluster"].(string) + if !ok { + sourceCluster = "" + } + delete(eventDetails, "source_cluster") + if err != nil { + eventDetails["error"] = err.Error() + eventDetails["error_type"] = util.ErrorType(err) + } + payload := wideevents.ReplicationLifecyclePayload{ + TaskType: wideevents.ReplTaskSyncWorkflowState, + Shard: shardContext.GetShardID(), + Namespace: namespaceName, + NamespaceID: request.GetNamespaceId(), + WorkflowID: request.GetParentExecution().GetWorkflowId(), + RunID: request.GetParentExecution().GetRunId(), + SourceCluster: sourceCluster, + } + switch outcome { + case wideevents.ParentChildOutcomeScheduled, + wideevents.ParentChildOutcomeStarted, + wideevents.ParentChildOutcomeDeduplicated: + eventDetails["operation"] = wideevents.ReplOperationStandbyVerificationSyncState + eventDetails["message"] = "Parent workflow resend checkpoint" + payload.Phase = wideevents.ReplicationExecuting + payload.Details = eventDetails + wideevents.Emit(shardContext.GetEventLogger(), payload) + case wideevents.ParentChildOutcomeSucceeded, wideevents.ParentChildOutcomeSourceNotFound: + eventDetails["operation"] = wideevents.ReplOperationStandbyVerificationSyncState + eventDetails["message"] = "Parent workflow resend checkpoint" + payload.Phase = wideevents.ReplicationApplied + payload.Outcome = wideevents.ParentChildOutcomeVerified + payload.Details = eventDetails + wideevents.Emit(shardContext.GetEventLogger(), payload) + default: + wideevents.EmitReplicationError( + shardContext.GetEventLogger(), + payload, + wideevents.ReplOperationStandbyVerificationSyncState, + "Parent workflow resend checkpoint", + err, + eventDetails, + ) + } +} diff --git a/service/history/history_engine2_test.go b/service/history/history_engine2_test.go index 6304ecc964..de14cf86c7 100644 --- a/service/history/history_engine2_test.go +++ b/service/history/history_engine2_test.go @@ -48,8 +48,11 @@ import ( "go.temporal.io/server/common/searchattribute/sadefs" serviceerrors "go.temporal.io/server/common/serviceerror" "go.temporal.io/server/common/tasktoken" + "go.temporal.io/server/common/testing/await" "go.temporal.io/server/common/testing/protorequire" "go.temporal.io/server/common/testing/testvars" + "go.temporal.io/server/common/util" + "go.temporal.io/server/common/wideevents" "go.temporal.io/server/service/history/api" "go.temporal.io/server/service/history/configs" "go.temporal.io/server/service/history/consts" @@ -89,6 +92,7 @@ type ( mockClusterMetadata *cluster.MockMetadata mockVisibilityManager *manager.MockVisibilityManager mockWorkflowStateReplicator *ndc.MockWorkflowStateReplicator + parentChildEventCapture *parentChildEventCapture workflowCache wcache.Cache historyEngine *historyEngineImpl @@ -142,6 +146,7 @@ func (s *engine2Suite) SetupTest() { s.mockMemoryScheduledQueue.EXPECT().NotifyNewTasks(gomock.Any()).AnyTimes() s.config = tests.NewDynamicConfig() + s.parentChildEventCapture = &parentChildEventCapture{} mockShard := shard.NewTestContext( s.controller, &persistencespb.ShardInfo{ @@ -149,6 +154,7 @@ func (s *engine2Suite) SetupTest() { RangeId: 1, }, s.config, + s.parentChildEventCapture, ) reg := hsm.NewRegistry() err := workflow.RegisterStateMachine(reg) @@ -2684,6 +2690,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 } + s.config.EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true) + capture := s.parentChildEventCapture request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{ NamespaceId: tests.ParentNamespaceID.String(), @@ -2705,7 +2713,7 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParent() 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) + mockClusterMetadata.EXPECT().GetAllClusterInfo().Return(cluster.TestAllClusterInfo).AnyTimes() mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(true, tests.Version).Return(cluster.TestCurrentClusterName).AnyTimes() s.mockShard.SetClusterMetadata(mockClusterMetadata) @@ -2740,17 +2748,13 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParent() ).Return(resp, nil) s.mockWorkflowStateReplicator.EXPECT().ReplicateVersionedTransition(gomock.Any(), chasm.WorkflowArchetypeID, resp.VersionedTransitionArtifact, cluster.TestCurrentClusterName).Return(nil) - // prepare closed workflow + // The parent can land as a zombie after conflict resolution; preserve that state in the event. ms := workflow.TestGlobalMutableState(s.historyEngine.shardContext, s.mockEventsCache, log.NewTestLogger(), tests.Version, tests.WorkflowID, tests.RunID) addWorkflowExecutionStartedEvent(ms, &commonpb.WorkflowExecution{ WorkflowId: tests.WorkflowID, RunId: tests.RunID, }, "wType", "testTaskQueue", payloads.EncodeString("input"), 25*time.Second, 20*time.Second, 200*time.Second, "identity") - _, err := ms.AddTimeoutWorkflowEvent( - enumspb.RETRY_STATE_RETRY_POLICY_NOT_SET, - uuid.NewString(), - ) - s.NoError(err) + ms.GetExecutionState().State = enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE ms.GetExecutionInfo().VersionHistories = &historyspb.VersionHistories{ CurrentVersionHistoryIndex: 0, Histories: []*historyspb.VersionHistory{ @@ -2767,14 +2771,79 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParent() gwmsResponse := &persistence.GetWorkflowExecutionResponse{State: wfMs} s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(gwmsResponse, nil) - _, err = s.historyEngine.VerifyChildExecutionCompletionRecorded(metrics.AddMetricsContext(context.Background()), request) - s.NoError(err) + _, err := s.historyEngine.VerifyChildExecutionCompletionRecorded(metrics.AddMetricsContext(context.Background()), request) + s.Require().NoError(err) + + s.Require().Equal([]string{ + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeSucceeded), + }, parentChildOutcomes(capture)) + records := parentChildRecords(capture) + s.Require().Len(records, 2) + attributes := wideEventAttributes(records[0]) + details := wideEventDetails(records[0]) + s.Equal(string(wideevents.ReplicationExecuting), attributes["phase"].AsString()) + s.Equal(wideevents.ParentChildPhaseParentResend, details["phase"]) + s.Equal(cluster.TestAlternativeClusterName, details["local_cluster"]) + s.Equal(int64(1), attributes["shard"].AsInt64()) + s.Equal(request.GetParentExecution().GetWorkflowId(), attributes["workflow_id"].AsString()) + s.Equal(request.GetParentExecution().GetWorkflowId(), details["parent_workflow_id"]) + s.Equal(request.GetChildExecution().GetWorkflowId(), details["child_workflow_id"]) + s.InDelta(float64(request.GetParentInitiatedId()), details["parent_initiated_id"], 0) + s.InDelta(float64(request.GetParentInitiatedVersion()), details["parent_initiated_version"], 0) + s.Equal(cluster.TestCurrentClusterName, attributes["source_cluster"].AsString()) + s.Equal(util.ErrorType(&serviceerror.NotFound{}), details["initial_error_type"]) + s.Equal("sync_workflow_state", details["stage"]) + s.Equal(string(wideevents.ReplicationApplied), wideEventAttributes(records[1])["phase"].AsString()) + s.Equal(wideevents.ParentChildOutcomeVerified, wideEventAttributes(records[1])["outcome"].AsString()) + s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE.String(), wideEventDetails(records[1])["parent_workflow_state"]) + + // Source NotFound is a terminal, successful outcome: there is no parent state left to pull. + sourceMissingRequest := &historyservice.VerifyChildExecutionCompletionRecordedRequest{ + NamespaceId: request.GetNamespaceId(), + ParentExecution: &commonpb.WorkflowExecution{ + WorkflowId: "parent missing on source", + RunId: uuid.NewString(), + }, + ChildExecution: request.GetChildExecution(), + ParentInitiatedId: request.GetParentInitiatedId(), + ParentInitiatedVersion: request.GetParentInitiatedVersion(), + ResendParent: true, + } + s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, &serviceerror.NotFound{}) + s.mockShard.Resource.RemoteAdminClient.EXPECT().SyncWorkflowState( + gomock.Any(), + &adminservice.SyncWorkflowStateRequest{ + NamespaceId: sourceMissingRequest.NamespaceId, + Execution: sourceMissingRequest.ParentExecution, + ArchetypeId: chasm.WorkflowArchetypeID, + TargetClusterId: int32( + cluster.TestAlternativeClusterInitialFailoverVersion, + ), + }, + ).Return(nil, serviceerror.NewNotFound("parent missing on source")) + + _, err = s.historyEngine.VerifyChildExecutionCompletionRecorded(metrics.AddMetricsContext(context.Background()), sourceMissingRequest) + s.Require().NoError(err) + s.Require().Equal([]string{ + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeSucceeded), + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeSourceNotFound), + }, parentChildOutcomes(capture)) + sourceNotFoundRecord := parentChildRecords(capture)[3] + s.Equal(string(wideevents.ReplicationApplied), wideEventAttributes(sourceNotFoundRecord)["phase"].AsString()) + s.Equal(wideevents.ParentChildOutcomeVerified, wideEventAttributes(sourceNotFoundRecord)["outcome"].AsString()) + sourceNotFoundDetails := wideEventDetails(sourceNotFoundRecord) + s.Equal(util.ErrorType(serviceerror.NewNotFound("")), sourceNotFoundDetails["error_type"]) } // 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 } + s.config.EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true) + capture := s.parentChildEventCapture request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{ NamespaceId: tests.ParentNamespaceID.String(), @@ -2819,12 +2888,60 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentAs case <-time.After(10 * time.Second): s.Fail("background resend did not call SyncWorkflowState") } + await.RequireTrue(s.T(), func() bool { + return len(parentChildOutcomes(capture)) >= 3 + }, 10*time.Second, 10*time.Millisecond) + s.Require().Equal([]string{ + string(wideevents.ParentChildOutcomeScheduled), + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeFailed), + }, parentChildOutcomes(capture)) + records := parentChildRecords(capture) + attributes := wideEventAttributes(records[2]) + details := wideEventDetails(records[2]) + s.Equal(util.ErrorType(serviceerror.NewUnavailable("source cluster unavailable")), details["error_type"]) + s.Equal(util.ErrorType(&serviceerror.NotFound{}), details["initial_error_type"]) + s.Equal("sync_workflow_state", details["stage"]) + s.Equal(cluster.TestCurrentClusterName, attributes["source_cluster"].AsString()) +} + +func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentLimited() { + s.config.EnableAsyncParentWorkflowResend = func() bool { return true } + s.config.ParentWorkflowResendMaxInFlight = func() int { return 0 } + s.config.EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true) + capture := s.parentChildEventCapture + + 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{}) + + _, err := s.historyEngine.VerifyChildExecutionCompletionRecorded(metrics.AddMetricsContext(context.Background()), request) + var notFound *serviceerror.NotFound + s.Require().ErrorAs(err, ¬Found) + s.Require().Equal([]string{string(wideevents.ParentChildOutcomeLimited)}, parentChildOutcomes(capture)) + details := wideEventDetails(parentChildRecords(capture)[0]) + s.Equal(util.ErrorType(&serviceerror.NotFound{}), details["initial_error_type"]) + s.InDelta(0, details["max_in_flight"], 0) } // 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 } + s.config.EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true) + capture := s.parentChildEventCapture request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{ NamespaceId: tests.ParentNamespaceID.String(), @@ -2853,6 +2970,12 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentDe // Times(1): only one resend may reach the source. It blocks so the second attempt overlaps. entered := make(chan struct{}) release := make(chan struct{}) + released := false + defer func() { + if !released { + close(release) + } + }() s.mockShard.Resource.RemoteAdminClient.EXPECT().SyncWorkflowState(gomock.Any(), gomock.Any()). DoAndReturn(func(context.Context, *adminservice.SyncWorkflowStateRequest, ...grpc.CallOption) (*adminservice.SyncWorkflowStateResponse, error) { close(entered) @@ -2873,8 +2996,25 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentDe // Second attempt while the first is in flight: must not start another resend. _, err = s.historyEngine.VerifyChildExecutionCompletionRecorded(ctx, request) s.Error(err) + s.Require().Equal([]string{ + string(wideevents.ParentChildOutcomeScheduled), + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeDeduplicated), + }, parentChildOutcomes(capture)) + deduplicatedRecord := parentChildRecords(capture)[2] + s.Equal(string(wideevents.ReplicationExecuting), wideEventAttributes(deduplicatedRecord)["phase"].AsString()) close(release) + released = true + await.RequireTrue(s.T(), func() bool { + return len(parentChildOutcomes(capture)) >= 4 + }, 10*time.Second, 10*time.Millisecond) + s.Require().Equal([]string{ + string(wideevents.ParentChildOutcomeScheduled), + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeDeduplicated), + string(wideevents.ParentChildOutcomeFailed), + }, parentChildOutcomes(capture)) } func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_WorkflowClosed() { diff --git a/service/history/parent_child_events.go b/service/history/parent_child_events.go new file mode 100644 index 0000000000..5f7dd2fa71 --- /dev/null +++ b/service/history/parent_child_events.go @@ -0,0 +1,205 @@ +package history + +import ( + "go.temporal.io/api/serviceerror" + "go.temporal.io/server/common/namespace" + "go.temporal.io/server/common/wideevents" + historyi "go.temporal.io/server/service/history/interfaces" + "go.temporal.io/server/service/history/tasks" +) + +func emitChildCompletionVerificationStarted( + shardContext historyi.ShardContext, + task *tasks.CloseExecutionTask, + parentNamespaceID string, + parentWorkflowID string, + parentRunID string, + parentInitiatedID int64, + parentInitiatedVersion int64, + childWorkflowState string, + resendParent bool, + attempt int, +) { + if !resendParent || !parentChildLifecycleEnabled(shardContext) { + return + } + payload, details := parentChildEventForCloseTask( + shardContext, + task, + parentNamespaceID, + parentWorkflowID, + parentRunID, + parentInitiatedID, + parentInitiatedVersion, + childWorkflowState, + attempt, + ) + details["phase"] = wideevents.ParentChildPhaseVerifyChildCompletion + details["outcome"] = wideevents.ParentChildOutcomeStarted + emitParentChildReplicationEvent( + shardContext, + payload, + wideevents.ReplicationExecuting, + wideevents.ReplOperationStandbyVerification, + "Standby child completion verification started with parent resend requested", + nil, + details, + ) +} + +func emitChildCompletionVerificationResult( + shardContext historyi.ShardContext, + task *tasks.CloseExecutionTask, + parentNamespaceID string, + parentWorkflowID string, + parentRunID string, + parentInitiatedID int64, + parentInitiatedVersion int64, + childWorkflowState string, + resendParent bool, + attempt int, + err error, +) { + if !resendParent || !parentChildLifecycleEnabled(shardContext) { + return + } + outcome, emit := parentChildVerificationOutcome( + err, + wideevents.ParentChildOutcomeNotFound, + wideevents.ParentChildOutcomeCompletionMissing, + ) + switch err.(type) { + case nil: + outcome, emit = wideevents.ParentChildOutcomeVerified, true + case *serviceerror.NamespaceNotFound, *serviceerror.Unimplemented: + outcome, emit = wideevents.ParentChildOutcomeIgnored, true + default: + } + if !emit { + return + } + payload, details := parentChildEventForCloseTask( + shardContext, + task, + parentNamespaceID, + parentWorkflowID, + parentRunID, + parentInitiatedID, + parentInitiatedVersion, + childWorkflowState, + attempt, + ) + details["phase"] = wideevents.ParentChildPhaseVerifyChildCompletion + details["outcome"] = outcome + if attempt >= 0 { + details["attempt"] = attempt + } + replicationPhase := wideevents.ReplicationError + if err == nil || outcome == wideevents.ParentChildOutcomeIgnored { + replicationPhase = wideevents.ReplicationApplied + } + emitParentChildReplicationEvent( + shardContext, + payload, + replicationPhase, + wideevents.ReplOperationStandbyVerification, + "Standby child completion verification completed", + err, + details, + ) +} + +func parentChildEventForCloseTask( + shardContext historyi.ShardContext, + task *tasks.CloseExecutionTask, + parentNamespaceID string, + parentWorkflowID string, + parentRunID string, + parentInitiatedID int64, + parentInitiatedVersion int64, + childWorkflowState string, + attempt int, +) (wideevents.ReplicationLifecyclePayload, map[string]any) { + payload, details := parentChildEventForTask(shardContext, task, attempt) + payload.ParentWorkflowID = parentWorkflowID + payload.ParentRunID = parentRunID + payload.ParentInitiatedID = parentInitiatedID + details["parent_namespace_id"] = parentNamespaceID + details["child_workflow_state"] = childWorkflowState + details["parent_initiated_version"] = parentInitiatedVersion + return payload, details +} + +func parentChildEventForTask( + shardContext historyi.ShardContext, + task tasks.Task, + attempt int, +) (wideevents.ReplicationLifecyclePayload, map[string]any) { + namespaceName := "" + if name, err := shardContext.GetNamespaceRegistry().GetNamespaceName(namespace.ID(task.GetNamespaceID())); err == nil { + namespaceName = name.String() + } + details := map[string]any{ + "event_type": wideevents.ParentChildLifecycleEventType, + "local_task_id": task.GetTaskID(), + } + if versionedTask, ok := task.(tasks.HasVersion); ok { + details["version"] = versionedTask.GetVersion() + } + return wideevents.ReplicationLifecyclePayload{ + TaskType: task.GetType().String(), + Shard: shardContext.GetShardID(), + Namespace: namespaceName, + NamespaceID: task.GetNamespaceID(), + WorkflowID: task.GetWorkflowID(), + RunID: task.GetRunID(), + Attempt: int32(attempt), + }, details +} + +func emitParentChildReplicationEvent( + shardContext historyi.ShardContext, + payload wideevents.ReplicationLifecyclePayload, + phase wideevents.ReplicationPhase, + operation string, + message string, + err error, + details map[string]any, +) { + if phase == wideevents.ReplicationError { + wideevents.EmitReplicationError(shardContext.GetEventLogger(), payload, operation, message, err, details) + return + } + details["operation"] = operation + details["message"] = message + payload.Phase = phase + payload.Details = details + if phase == wideevents.ReplicationApplied { + payload.Outcome = wideevents.ParentChildOutcomeVerified + } + wideevents.Emit(shardContext.GetEventLogger(), payload) +} + +func parentChildLifecycleEnabled(shardContext historyi.ShardContext) bool { + return shardContext.GetConfig().EmitReplicationLifecycleEvents() +} + +func parentChildVerificationOutcome( + err error, + notFoundOutcome string, + notReadyOutcome string, +) (string, bool) { + if err == nil { + return "", false + } + switch err.(type) { + case *serviceerror.NotFound: + return notFoundOutcome, true + case *serviceerror.WorkflowNotReady: + return notReadyOutcome, true + case *serviceerror.NamespaceNotFound, *serviceerror.Unimplemented: + return "", false + default: + return wideevents.ParentChildOutcomeFailed, true + } +} diff --git a/service/history/parent_child_events_test.go b/service/history/parent_child_events_test.go new file mode 100644 index 0000000000..d31adb8285 --- /dev/null +++ b/service/history/parent_child_events_test.go @@ -0,0 +1,80 @@ +package history + +import ( + "context" + "encoding/json" + "sync" + + otellog "go.opentelemetry.io/otel/log" + "go.opentelemetry.io/otel/log/embedded" + "go.temporal.io/server/common/wideevents" +) + +type parentChildEventCapture struct { + embedded.Logger + + mu sync.RWMutex + records []otellog.Record +} + +func (c *parentChildEventCapture) Emit(_ context.Context, record otellog.Record) { + c.mu.Lock() + defer c.mu.Unlock() + c.records = append(c.records, record.Clone()) +} + +func (*parentChildEventCapture) Enabled(context.Context, otellog.EnabledParameters) bool { + return true +} + +func (c *parentChildEventCapture) snapshot() []otellog.Record { + c.mu.RLock() + defer c.mu.RUnlock() + records := make([]otellog.Record, len(c.records)) + for i, record := range c.records { + records[i] = record.Clone() + } + return records +} + +func wideEventAttributes(record otellog.Record) map[string]otellog.Value { + attributes := make(map[string]otellog.Value, record.AttributesLen()) + record.WalkAttributes(func(attribute otellog.KeyValue) bool { + attributes[attribute.Key] = attribute.Value + return true + }) + return attributes +} + +func parentChildOutcomes(capture *parentChildEventCapture) []string { + var outcomes []string + for _, record := range parentChildRecords(capture) { + if outcome, ok := wideEventDetails(record)["outcome"].(string); ok { + outcomes = append(outcomes, outcome) + } + } + return outcomes +} + +func parentChildRecords(capture *parentChildEventCapture) []otellog.Record { + var records []otellog.Record + for _, record := range capture.snapshot() { + if record.EventName() == wideevents.ReplicationLifecycleEventName && + wideEventDetails(record)["event_type"] == wideevents.ParentChildLifecycleEventType { + records = append(records, record) + } + } + return records +} + +func wideEventDetails(record otellog.Record) map[string]any { + attribute, ok := wideEventAttributes(record)["details"] + if !ok { + return nil + } + var details map[string]any + if err := json.Unmarshal([]byte(attribute.AsString()), &details); err != nil { + return nil + } + return details +} diff --git a/service/history/shard/context_testutil.go b/service/history/shard/context_testutil.go index 7ef7ecbefa..375b4ade90 100644 --- a/service/history/shard/context_testutil.go +++ b/service/history/shard/context_testutil.go @@ -5,6 +5,7 @@ import ( "fmt" "sync" + otellog "go.opentelemetry.io/otel/log" "go.temporal.io/server/api/historyservice/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/chasm" @@ -44,8 +45,9 @@ func NewTestContextWithTimeSource( shardInfo *persistencespb.ShardInfo, config *configs.Config, timeSource clock.TimeSource, + eventLogger ...otellog.Logger, ) *ContextTest { - result := NewTestContext(ctrl, shardInfo, config) + result := NewTestContext(ctrl, shardInfo, config, eventLogger...) result.timeSource = timeSource result.taskKeyManager.generator.timeSource = timeSource result.Resource.TimeSource = timeSource @@ -56,15 +58,21 @@ func NewTestContext( ctrl *gomock.Controller, shardInfo *persistencespb.ShardInfo, config *configs.Config, + eventLogger ...otellog.Logger, ) *ContextTest { + var logger otellog.Logger + if len(eventLogger) > 0 { + logger = eventLogger[0] + } resourceTest := resourcetest.NewTest(ctrl, primitives.HistoryService) eventsCache := events.NewMockCache(ctrl) shard := newTestContext( resourceTest, eventsCache, ContextConfigOverrides{ - ShardInfo: shardInfo, - Config: config, + ShardInfo: shardInfo, + Config: config, + EventLogger: logger, }, ) return &ContextTest{ @@ -80,6 +88,7 @@ type ContextConfigOverrides struct { Registry namespace.Registry ClusterMetadata cluster.Metadata ExecutionManager persistence.ExecutionManager + EventLogger otellog.Logger } type StubContext struct { @@ -141,6 +150,7 @@ func newTestContext(t *resourcetest.Test, eventsCache events.Cache, config Conte lifecycleCtx: lifecycleCtx, lifecycleCancel: lifecycleCancel, queueMetricEmitter: sync.Once{}, + eventLogger: config.EventLogger, state: contextStateAcquired, engineFuture: future.NewFuture[historyi.Engine](), diff --git a/service/history/transfer_queue_standby_task_executor.go b/service/history/transfer_queue_standby_task_executor.go index 9e0f1df07f..e8c309d382 100644 --- a/service/history/transfer_queue_standby_task_executor.go +++ b/service/history/transfer_queue_standby_task_executor.go @@ -103,7 +103,7 @@ func (t *transferQueueStandbyTaskExecutor) Execute( // TODO: add error logs err = nil case *tasks.CloseExecutionTask: - err = t.processCloseExecution(ctx, task) + err = t.processCloseExecution(ctx, task, executable.Attempt()) case *tasks.DeleteExecutionTask: err = t.processDeleteExecutionTask(ctx, task, false) case *tasks.ChasmTask: @@ -294,6 +294,7 @@ func (t *transferQueueStandbyTaskExecutor) processWorkflowTask( func (t *transferQueueStandbyTaskExecutor) processCloseExecution( ctx context.Context, transferTask *tasks.CloseExecutionTask, + attempt int, ) error { processTaskIfClosed := true actionFn := func(ctx context.Context, wfContext historyi.WorkflowContext, mutableState historyi.MutableState, release historyi.ReleaseWorkflowContextFunc) (any, error) { @@ -339,10 +340,24 @@ func (t *transferQueueStandbyTaskExecutor) processCloseExecution( parentInitiatedID := executionInfo.ParentInitiatedId parentInitiatedVersion := executionInfo.ParentInitiatedVersion parentClock := executionInfo.ParentClock + childWorkflowState := mutableState.GetExecutionState().GetState().String() // no need for mutable state anymore, release workflow lock release(nil) + emitChildCompletionVerificationStarted( + t.shardContext, + transferTask, + parentNamespaceID, + parentWorkflowID, + parentRunID, + parentInitiatedID, + parentInitiatedVersion, + childWorkflowState, + resendParent, + attempt, + ) + _, err := t.historyRawClient.VerifyChildExecutionCompletionRecorded(ctx, &historyservice.VerifyChildExecutionCompletionRecordedRequest{ NamespaceId: parentNamespaceID, ParentExecution: &commonpb.WorkflowExecution{ @@ -358,6 +373,19 @@ func (t *transferQueueStandbyTaskExecutor) processCloseExecution( Clock: parentClock, ResendParent: resendParent, }) + emitChildCompletionVerificationResult( + t.shardContext, + transferTask, + parentNamespaceID, + parentWorkflowID, + parentRunID, + parentInitiatedID, + parentInitiatedVersion, + childWorkflowState, + resendParent, + attempt, + err, + ) switch err.(type) { case nil, *serviceerror.NamespaceNotFound, *serviceerror.Unimplemented: // Case 1: Target workflow is in the desired state. @@ -491,7 +519,6 @@ func (t *transferQueueStandbyTaskExecutor) processStartChildExecution( childStartedWorkflowID := childWorkflowInfo.StartedWorkflowId childStartedRunID := childWorkflowInfo.StartedRunId childClock := childWorkflowInfo.Clock - // no need for mutable state anymore, release workflow lock release(nil) diff --git a/service/history/transfer_queue_standby_task_executor_test.go b/service/history/transfer_queue_standby_task_executor_test.go index ab7d622150..81729905e2 100644 --- a/service/history/transfer_queue_standby_task_executor_test.go +++ b/service/history/transfer_queue_standby_task_executor_test.go @@ -30,6 +30,7 @@ import ( "go.temporal.io/server/common/clock" "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/definition" + "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/namespace" @@ -41,6 +42,8 @@ import ( "go.temporal.io/server/common/telemetry" "go.temporal.io/server/common/testing/mockapi/workflowservicemock/v1" "go.temporal.io/server/common/testing/protomock" + "go.temporal.io/server/common/util" + "go.temporal.io/server/common/wideevents" "go.temporal.io/server/service/history/consts" "go.temporal.io/server/service/history/events" "go.temporal.io/server/service/history/hsm" @@ -76,9 +79,10 @@ type ( mockRemoteAdminClient *adminservicemock.MockAdminServiceClient mockChasmEngine chasm.Engine - mockExecutionMgr *persistence.MockExecutionManager - mockArchivalMetadata archiver.MetadataMock - mockArchiverProvider *provider.MockArchiverProvider + mockExecutionMgr *persistence.MockExecutionManager + mockArchivalMetadata archiver.MetadataMock + mockArchiverProvider *provider.MockArchiverProvider + parentChildEventCapture *parentChildEventCapture workflowCache wcache.Cache logger log.Logger @@ -124,6 +128,7 @@ func (s *transferQueueStandbyTaskExecutorSuite) SetupTest() { s.chasmDiscardDuration = config.ChasmStandbyTaskDiscardDelay("") s.controller = gomock.NewController(s.T()) + s.parentChildEventCapture = &parentChildEventCapture{} s.mockShard = shard.NewTestContextWithTimeSource( s.controller, &persistencespb.ShardInfo{ @@ -131,6 +136,7 @@ func (s *transferQueueStandbyTaskExecutorSuite) SetupTest() { }, config, s.timeSource, + s.parentChildEventCapture, ) reg := hsm.NewRegistry() @@ -725,6 +731,9 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessWorkflowTask_StampMis } func (s *transferQueueStandbyTaskExecutorSuite) TestProcessCloseExecution() { + capture := s.parentChildEventCapture + s.mockShard.GetConfig().EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true) + execution := &commonpb.WorkflowExecution{ WorkflowId: "some random workflow ID", RunId: uuid.NewString(), @@ -841,8 +850,17 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessCloseExecution() { s.ErrorAs(resp.ExecutionErr, &verificationErr) var resourceExhaustedErr *serviceerror.ResourceExhausted s.ErrorAs(resp.ExecutionErr, &resourceExhaustedErr) + s.Require().Empty(parentChildRecords(capture)) s.mockShard.SetCurrentTime(s.clusterName, now.Add(s.localVerificationDuration)) + s.mockHistoryClient.EXPECT().VerifyChildExecutionCompletionRecorded(gomock.Any(), expectedVerificationWithResendParentRequest).Return(nil, nil) + resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask)) + s.NoError(resp.ExecutionErr) + + s.mockHistoryClient.EXPECT().VerifyChildExecutionCompletionRecorded(gomock.Any(), expectedVerificationWithResendParentRequest).Return(nil, serviceerror.NewUnimplemented("not implemented")) + resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask)) + s.NoError(resp.ExecutionErr) + s.mockHistoryClient.EXPECT().VerifyChildExecutionCompletionRecorded(gomock.Any(), expectedVerificationWithResendParentRequest).Return(nil, consts.ErrWorkflowNotReady) resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask)) s.Equal(consts.ErrTaskRetry, resp.ExecutionErr) @@ -862,6 +880,68 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessCloseExecution() { resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask)) s.ErrorAs(resp.ExecutionErr, &verificationErr) s.Equal(randomErr, verificationErr.Unwrap()) + + s.Equal([]string{ + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeVerified), + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeIgnored), + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeCompletionMissing), + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeCompletionMissing), + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeCompletionMissing), + string(wideevents.ParentChildOutcomeStarted), + string(wideevents.ParentChildOutcomeFailed), + }, parentChildOutcomes(capture)) + records := parentChildRecords(capture) + s.Require().Len(records, 12) + attributes := wideEventAttributes(records[0]) + details := wideEventDetails(records[0]) + s.Equal(string(wideevents.ReplicationExecuting), attributes["phase"].AsString()) + s.Equal(wideevents.ParentChildPhaseVerifyChildCompletion, details["phase"]) + s.Equal(wideevents.ParentChildOutcomeStarted, details["outcome"]) + s.Equal(s.namespaceID.String(), attributes["namespace_id"].AsString()) + s.Equal(parentExecution.GetWorkflowId(), attributes["parent_workflow_id"].AsString()) + s.Equal(parentExecution.GetRunId(), attributes["parent_run_id"].AsString()) + s.Equal(execution.GetWorkflowId(), attributes["workflow_id"].AsString()) + s.Equal(execution.GetRunId(), attributes["run_id"].AsString()) + s.Equal(transferTask.GetType().String(), attributes["task_type"].AsString()) + s.Equal(int64(1), attributes["attempt"].AsInt64()) + s.Equal("Completed", details["child_workflow_state"]) + s.Equal(parentInitiatedID, attributes["parent_initiated_id"].AsInt64()) + s.Equal(parentNamespaceID, details["parent_namespace_id"]) + s.InDelta(float64(parentInitiatedVersion), details["parent_initiated_version"], 0) + s.InDelta(float64(taskID), details["local_task_id"], 0) + s.InDelta(float64(transferTask.GetVersion()), details["version"], 0) + s.Require().NotContains(details, "attempt") + for _, key := range []string{ + "child_namespace_id", + "child_workflow_id", + "child_run_id", + "local_cluster", + "local_task_type", + "parent_initiated_id", + "parent_run_id", + "parent_workflow_id", + "resend_parent_requested", + "verification_scope", + } { + for _, record := range records { + s.Require().NotContains(wideEventDetails(record), key) + } + } + missingDetails := wideEventDetails(records[5]) + s.Equal(wideevents.ParentChildOutcomeCompletionMissing, missingDetails["outcome"]) + s.Equal(util.ErrorType(consts.ErrWorkflowNotReady), missingDetails["error_type"]) + s.InDelta(1, missingDetails["attempt"], 0) + verifiedAttributes := wideEventAttributes(records[1]) + s.Require().NotContains(verifiedAttributes, "attempt") + s.InDelta(1, wideEventDetails(records[1])["attempt"], 0) + ignoredAttributes := wideEventAttributes(records[3]) + s.Equal(string(wideevents.ReplicationApplied), ignoredAttributes["phase"].AsString()) + s.Equal(wideevents.ParentChildOutcomeVerified, ignoredAttributes["outcome"].AsString()) } func (s *transferQueueStandbyTaskExecutorSuite) TestProcessCancelExecution_Pending() { @@ -1125,6 +1205,9 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessSignalExecution_Succe } func (s *transferQueueStandbyTaskExecutorSuite) TestProcessStartChildExecution_Pending() { + capture := s.parentChildEventCapture + s.mockShard.GetConfig().EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true) + execution := &commonpb.WorkflowExecution{ WorkflowId: "some random workflow ID", RunId: uuid.NewString(), @@ -1185,7 +1268,8 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessStartChildExecution_P resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask)) s.Equal(consts.ErrTaskRetry, resp.ExecutionErr) - event = addChildWorkflowExecutionStartedEvent(mutableState, event.GetEventId(), childWorkflowID, uuid.NewString(), childWorkflowType, nil) + childRunID := uuid.NewString() + event = addChildWorkflowExecutionStartedEvent(mutableState, event.GetEventId(), childWorkflowID, childRunID, childWorkflowType, nil) mutableState.FlushBufferedEvents() // clear the cache @@ -1230,6 +1314,8 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessStartChildExecution_P s.mockHistoryClient.EXPECT().VerifyFirstWorkflowTaskScheduled(gomock.Any(), gomock.Any()).Return(nil, nil) resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask)) s.NoError(resp.ExecutionErr) + + s.Require().Empty(parentChildRecords(capture)) } func (s *transferQueueStandbyTaskExecutorSuite) TestProcessStartChildExecution_Success() {