replication: emit lifecycle error events

[michaely520]

What changed?

• Adds phase=error to the existing replication_lifecycle wide event; no new event type or table.
• Covers state-based replication (SyncVersionedTransition, VerifyVersionedTransition, and SyncWorkflowState) plus standby transfer, timer, and outbound queue failures.
• Captures sender, passive execution/apply, verification, recovery/refetch, namespace refresh, Nack, DLQ, and history-branch cleanup boundaries.
• Uses one shared error builder with small sender, executable-task, NDC, and standby-queue adapters.
• Records workflow identity, source task identity, target context, operation, error, attempt/priority, disposition/recovery, and extensible diagnostics in details.
• Identifies apply provenance as apply_artifact_source=task_payload or sync_state_refetch.
• Remains gated by history.emitReplicationLifecycleEvents (default off).

Why?

Replication failures span the sender, passive executor, recovery loop, and apply layer. Recording these boundaries in the existing lifecycle event makes the path of a workflow or replication task directly traceable without adding another event schema.

Example traces

The examples below are abridged records captured from the two-cluster XDC test. Events for the same task correlate on source_cluster, source_shard, and source_task_id; workflow identity is present on every record.

A state task that fails on the passive cluster and is written to the DLQ:

{"phase":"sent","task_type":"sync_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048581,"priority":"High"}
{"phase":"error","task_type":"sync_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048581,"details":{"operation":"passive_task_execution","error":"failed to apply replication task","error_type":"serviceerror.InvalidArgument","apply_artifact_source":"task_payload","attempt":1,"priority":"High","target_cluster":"standby"}}
{"phase":"error","task_type":"sync_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048581,"details":{"operation":"task_execution","error":"failed to apply replication task","terminal":true,"priority":"High","target_cluster":"standby"}}
{"phase":"error","task_type":"sync_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048581,"details":{"operation":"dlq_write","disposition":"dlq","terminal":true,"priority":"High","target_cluster":"standby","target_shard":1}}

A verification task that detects missing state, refetches it, and then verifies successfully:

{"phase":"sent","task_type":"verify_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048595,"priority":"High"}
{"phase":"executing","task_type":"verify_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048595,"attempt":1}
{"phase":"applied","task_type":"verify_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048595,"outcome":"resend_needed"}
{"phase":"error","task_type":"verify_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048595,"details":{"operation":"standby_verification","error":"missing mutable state, resend","error_type":"serviceerror.SyncState","recovery_action":"sync_state","priority":"High","target_cluster":"standby"}}
{"phase":"applied","task_type":"sync_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048595,"outcome":"applied","details":{"apply_artifact_source":"sync_state_refetch"}}
{"phase":"executing","task_type":"verify_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048595,"attempt":1}
{"phase":"applied","task_type":"verify_versioned_transition","workflow_id":"example-workflow","source_cluster":"active","source_shard":1,"source_task_id":1048595,"outcome":"verified"}

How was it tested?

• go test -tags test_dep ./common/wideevents ./service/history/replication ./service/history/ndc ./service/history ./tests/testcore
• Changed-lines golangci-lint: 0 issues.
• Temporary, uncommitted two-cluster XDC tests forced passive task failures, DLQ handling, standby verification, and SyncState refetch/recovery with lifecycle events both enabled and disabled.
• A tiered-processing run confirmed concrete High priority on sent and error records.

Risks

• Enabling the dynamic config increases event volume; retries and recovery can produce several error phases for one source task.
• details.operation, details.disposition, and details.recovery_action distinguish those boundaries.
• Emission is best effort and does not change replication error propagation, retry, or recovery behavior.

[yux0]

can we check this inside emitReplicationApplyError or event logger?

[michaely520]

i consolidated into the emit fn

[qyc5937]

I think this would log duplicate as well which is fine but in many paths we're treating duplicates as success. For example

temporal/service/history/replication/executable_sync_versioned_transition_task.go

Lines 136 to 138 in d83b12e

	if errors.Is(err, consts.ErrDuplicate) {
	e.MarkTaskDuplicated()
	return nil

. not sure if it's worth converging.

[michaely520]

I removed this duplicate check so that we can have visibility into truly what happened here for the event - I'd rather see the duplicate than not see it and wonder if something fell into a black hole or not, duplicate is still an outcome

[qyc5937]

I think this would capture the retry errors on the standby as well. If volume is a concern, we could filter it out and just do the terminal errors.

[michaely520]

for now this volume isnt a concern, most of the time we dont see many retries