From 06e5531513f013a846844507250fba9860826959 Mon Sep 17 00:00:00 2001 From: Lina Jodoin Date: Thu, 13 Aug 2026 14:51:11 -0700 Subject: [PATCH] [CHASM] Support WithRequestID on UpdateComponent (#11169) ## What changed? - `WithRequestID` now applies to `UpdateComponent`, enabling execution-level idempotency guarding via request ID. - When a request ID is passed as part of an `UpdateComponent` call (via API handler), it is persisted upon successful updateFn call. If it is already present, instead, `UpdateComponent` fails with a `FailedPrecondition`. - When a request ID is not passed in, a generated ID is still created for error tracing purposes, but it is not written to mutable state. - On transaction close, mutable state will sweep the oldest RequestIDs (with an `attach_time`) upon hitting the configured limit. - This will sweep both entries below a configurable max age, as well as past a certain hard length limit. ## Why? - Scheduler's `UpdateSchedule` and `PatchSchedule` are implemented as handlers that persist a signal in V1. V1 signals provide idempotency via their request IDs. Scheduler V2 doesn't make use of signals, so instead, it must record request IDs explicitly. - We reuse the existing map within mutable state. - We *must* fail with an explicit error (`FailedPrecondition`) instead of simply returning a zero value (as Signals would on repeated successful requests). This is because `UpdateComponent` can apply to API models that include response values (which we don't record, therefore, we can't return on subsequent calls). ## How did you test it? - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s) --------- Co-authored-by: Fred Tzeng --- api/persistence/v1/executions.pb.go | 238 ++++++++------- chasm/chasmtest/test_engine.go | 63 ++-- chasm/chasmtest/test_engine_test.go | 87 +++++- chasm/engine.go | 16 +- common/dynamicconfig/constants.go | 12 + common/metrics/metric_defs.go | 4 + .../api/persistence/v1/executions.proto | 10 +- service/history/chasm_engine.go | 40 ++- service/history/chasm_engine_test.go | 224 ++++++++++++++ service/history/configs/config.go | 4 + service/history/interfaces/mutable_state.go | 3 + .../history/interfaces/mutable_state_mock.go | 12 + .../history/workflow/mutable_state_impl.go | 110 ++++++- .../workflow/mutable_state_impl_test.go | 286 ++++++++++++++++++ 14 files changed, 966 insertions(+), 143 deletions(-) diff --git a/api/persistence/v1/executions.pb.go b/api/persistence/v1/executions.pb.go index 260c766c87..ce4ad9f364 100644 --- a/api/persistence/v1/executions.pb.go +++ b/api/persistence/v1/executions.pb.go @@ -1444,8 +1444,11 @@ type WorkflowExecutionState struct { LastUpdateVersionedTransition *VersionedTransition `protobuf:"bytes,5,opt,name=last_update_versioned_transition,json=lastUpdateVersionedTransition,proto3" json:"last_update_versioned_transition,omitempty"` StartTime *timestamppb.Timestamp `protobuf:"bytes,6,opt,name=start_time,json=startTime,proto3" json:"start_time,omitempty"` // Request IDs that are attached to the workflow execution. It can be the request ID that started - // the workflow execution or request IDs that were attached to an existing running workflow - // execution via StartWorkflowExecutionRequest.OnConflictOptions. + // the workflow execution, request IDs that were attached to an existing running workflow + // execution via StartWorkflowExecutionRequest.OnConflictOptions, or (for CHASM executions) request + // IDs recorded for UpdateComponent idempotency - the latter are marked by RequestIDInfo.attach_time + // and are swept per-execution by count (history.maximumRequestIDsPerExecution) and age + // (history.requestIDMaxAge). RequestIds map[string]*RequestIDInfo `protobuf:"bytes,7,rep,name=request_ids,json=requestIds,proto3" json:"request_ids,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` // Run ID of the first execution in the chain (set on WorkflowExecutionStarted). Equals run_id // only for a first run; any continuation (continue-as-new, retry, cron, or reset) @@ -1543,9 +1546,12 @@ func (x *WorkflowExecutionState) GetFirstExecutionRunId() string { } type RequestIDInfo struct { - state protoimpl.MessageState `protogen:"open.v1"` - EventType v11.EventType `protobuf:"varint,1,opt,name=event_type,json=eventType,proto3,enum=temporal.api.enums.v1.EventType" json:"event_type,omitempty"` - EventId int64 `protobuf:"varint,2,opt,name=event_id,json=eventId,proto3" json:"event_id,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + EventType v11.EventType `protobuf:"varint,1,opt,name=event_type,json=eventType,proto3,enum=temporal.api.enums.v1.EventType" json:"event_type,omitempty"` + EventId int64 `protobuf:"varint,2,opt,name=event_id,json=eventId,proto3" json:"event_id,omitempty"` + // Set only for request IDs attached by the CHASM framework for UpdateComponent idempotency. + // Used as an ordering key for lazy eviction when set. + AttachTime *timestamppb.Timestamp `protobuf:"bytes,3,opt,name=attach_time,json=attachTime,proto3" json:"attach_time,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1594,6 +1600,13 @@ func (x *RequestIDInfo) GetEventId() int64 { return 0 } +func (x *RequestIDInfo) GetAttachTime() *timestamppb.Timestamp { + if x != nil { + return x.AttachTime + } + return nil +} + // transfer column type TransferTaskInfo struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -5120,11 +5133,13 @@ const file_temporal_server_api_persistence_v1_executions_proto_rawDesc = "" + "\x16first_execution_run_id\x18\b \x01(\tR\x13firstExecutionRunId\x1ap\n" + "\x0fRequestIdsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12G\n" + - "\x05value\x18\x02 \x01(\v21.temporal.server.api.persistence.v1.RequestIDInfoR\x05value:\x028\x01\"k\n" + + "\x05value\x18\x02 \x01(\v21.temporal.server.api.persistence.v1.RequestIDInfoR\x05value:\x028\x01\"\xa8\x01\n" + "\rRequestIDInfo\x12?\n" + "\n" + "event_type\x18\x01 \x01(\x0e2 .temporal.api.enums.v1.EventTypeR\teventType\x12\x19\n" + - "\bevent_id\x18\x02 \x01(\x03R\aeventId\"\xdf\a\n" + + "\bevent_id\x18\x02 \x01(\x03R\aeventId\x12;\n" + + "\vattach_time\x18\x03 \x01(\v2\x1a.google.protobuf.TimestampR\n" + + "attachTime\"\xdf\a\n" + "\x10TransferTaskInfo\x12!\n" + "\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12\x1f\n" + "\vworkflow_id\x18\x02 \x01(\tR\n" + @@ -5607,110 +5622,111 @@ var file_temporal_server_api_persistence_v1_executions_proto_depIdxs = []int32{ 47, // 58: temporal.server.api.persistence.v1.WorkflowExecutionState.start_time:type_name -> google.protobuf.Timestamp 37, // 59: temporal.server.api.persistence.v1.WorkflowExecutionState.request_ids:type_name -> temporal.server.api.persistence.v1.WorkflowExecutionState.RequestIdsEntry 68, // 60: temporal.server.api.persistence.v1.RequestIDInfo.event_type:type_name -> temporal.api.enums.v1.EventType - 69, // 61: temporal.server.api.persistence.v1.TransferTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType - 47, // 62: temporal.server.api.persistence.v1.TransferTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp - 38, // 63: temporal.server.api.persistence.v1.TransferTaskInfo.close_execution_task_details:type_name -> temporal.server.api.persistence.v1.TransferTaskInfo.CloseExecutionTaskDetails - 70, // 64: temporal.server.api.persistence.v1.TransferTaskInfo.chasm_task_info:type_name -> temporal.server.api.persistence.v1.ChasmTaskInfo - 69, // 65: temporal.server.api.persistence.v1.ReplicationTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType - 47, // 66: temporal.server.api.persistence.v1.ReplicationTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp - 71, // 67: temporal.server.api.persistence.v1.ReplicationTaskInfo.priority:type_name -> temporal.server.api.enums.v1.TaskPriority - 56, // 68: temporal.server.api.persistence.v1.ReplicationTaskInfo.versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 9, // 69: temporal.server.api.persistence.v1.ReplicationTaskInfo.task_equivalents:type_name -> temporal.server.api.persistence.v1.ReplicationTaskInfo - 72, // 70: temporal.server.api.persistence.v1.ReplicationTaskInfo.last_version_history_item:type_name -> temporal.server.api.history.v1.VersionHistoryItem - 69, // 71: temporal.server.api.persistence.v1.VisibilityTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType - 47, // 72: temporal.server.api.persistence.v1.VisibilityTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp - 47, // 73: temporal.server.api.persistence.v1.VisibilityTaskInfo.close_time:type_name -> google.protobuf.Timestamp - 47, // 74: temporal.server.api.persistence.v1.VisibilityTaskInfo.start_time:type_name -> google.protobuf.Timestamp - 70, // 75: temporal.server.api.persistence.v1.VisibilityTaskInfo.chasm_task_info:type_name -> temporal.server.api.persistence.v1.ChasmTaskInfo - 69, // 76: temporal.server.api.persistence.v1.TimerTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType - 62, // 77: temporal.server.api.persistence.v1.TimerTaskInfo.timeout_type:type_name -> temporal.api.enums.v1.TimeoutType - 73, // 78: temporal.server.api.persistence.v1.TimerTaskInfo.workflow_backoff_type:type_name -> temporal.server.api.enums.v1.WorkflowBackoffType - 47, // 79: temporal.server.api.persistence.v1.TimerTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp - 70, // 80: temporal.server.api.persistence.v1.TimerTaskInfo.chasm_task_info:type_name -> temporal.server.api.persistence.v1.ChasmTaskInfo - 56, // 81: temporal.server.api.persistence.v1.TimerTaskInfo.versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 69, // 82: temporal.server.api.persistence.v1.ArchivalTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType - 47, // 83: temporal.server.api.persistence.v1.ArchivalTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp - 69, // 84: temporal.server.api.persistence.v1.OutboundTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType - 47, // 85: temporal.server.api.persistence.v1.OutboundTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp - 74, // 86: temporal.server.api.persistence.v1.OutboundTaskInfo.state_machine_info:type_name -> temporal.server.api.persistence.v1.StateMachineTaskInfo - 70, // 87: temporal.server.api.persistence.v1.OutboundTaskInfo.chasm_task_info:type_name -> temporal.server.api.persistence.v1.ChasmTaskInfo - 14, // 88: temporal.server.api.persistence.v1.OutboundTaskInfo.worker_commands_task:type_name -> temporal.server.api.persistence.v1.WorkerCommandsTask - 75, // 89: temporal.server.api.persistence.v1.WorkerCommandsTask.commands:type_name -> temporal.api.worker.v1.WorkerCommand - 47, // 90: temporal.server.api.persistence.v1.ActivityInfo.scheduled_time:type_name -> google.protobuf.Timestamp - 47, // 91: temporal.server.api.persistence.v1.ActivityInfo.started_time:type_name -> google.protobuf.Timestamp - 48, // 92: temporal.server.api.persistence.v1.ActivityInfo.schedule_to_start_timeout:type_name -> google.protobuf.Duration - 48, // 93: temporal.server.api.persistence.v1.ActivityInfo.schedule_to_close_timeout:type_name -> google.protobuf.Duration - 48, // 94: temporal.server.api.persistence.v1.ActivityInfo.start_to_close_timeout:type_name -> google.protobuf.Duration - 48, // 95: temporal.server.api.persistence.v1.ActivityInfo.heartbeat_timeout:type_name -> google.protobuf.Duration - 48, // 96: temporal.server.api.persistence.v1.ActivityInfo.retry_initial_interval:type_name -> google.protobuf.Duration - 48, // 97: temporal.server.api.persistence.v1.ActivityInfo.retry_maximum_interval:type_name -> google.protobuf.Duration - 47, // 98: temporal.server.api.persistence.v1.ActivityInfo.retry_expiration_time:type_name -> google.protobuf.Timestamp - 76, // 99: temporal.server.api.persistence.v1.ActivityInfo.retry_last_failure:type_name -> temporal.api.failure.v1.Failure - 77, // 100: temporal.server.api.persistence.v1.ActivityInfo.last_heartbeat_details:type_name -> temporal.api.common.v1.Payloads - 47, // 101: temporal.server.api.persistence.v1.ActivityInfo.last_heartbeat_update_time:type_name -> google.protobuf.Timestamp - 78, // 102: temporal.server.api.persistence.v1.ActivityInfo.activity_type:type_name -> temporal.api.common.v1.ActivityType - 39, // 103: temporal.server.api.persistence.v1.ActivityInfo.use_workflow_build_id_info:type_name -> temporal.server.api.persistence.v1.ActivityInfo.UseWorkflowBuildIdInfo - 55, // 104: temporal.server.api.persistence.v1.ActivityInfo.last_worker_version_stamp:type_name -> temporal.api.common.v1.WorkerVersionStamp - 56, // 105: temporal.server.api.persistence.v1.ActivityInfo.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 47, // 106: temporal.server.api.persistence.v1.ActivityInfo.first_scheduled_time:type_name -> google.protobuf.Timestamp - 47, // 107: temporal.server.api.persistence.v1.ActivityInfo.last_attempt_complete_time:type_name -> google.protobuf.Timestamp - 79, // 108: temporal.server.api.persistence.v1.ActivityInfo.last_started_deployment:type_name -> temporal.api.deployment.v1.Deployment - 65, // 109: temporal.server.api.persistence.v1.ActivityInfo.last_deployment_version:type_name -> temporal.api.deployment.v1.WorkerDeploymentVersion - 60, // 110: temporal.server.api.persistence.v1.ActivityInfo.priority:type_name -> temporal.api.common.v1.Priority - 40, // 111: temporal.server.api.persistence.v1.ActivityInfo.pause_info:type_name -> temporal.server.api.persistence.v1.ActivityInfo.PauseInfo - 53, // 112: temporal.server.api.persistence.v1.ActivityInfo.started_clock:type_name -> temporal.server.api.clock.v1.VectorClock - 47, // 113: temporal.server.api.persistence.v1.TimerInfo.expiry_time:type_name -> google.protobuf.Timestamp - 56, // 114: temporal.server.api.persistence.v1.TimerInfo.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 80, // 115: temporal.server.api.persistence.v1.ChildExecutionInfo.parent_close_policy:type_name -> temporal.api.enums.v1.ParentClosePolicy - 53, // 116: temporal.server.api.persistence.v1.ChildExecutionInfo.clock:type_name -> temporal.server.api.clock.v1.VectorClock - 56, // 117: temporal.server.api.persistence.v1.ChildExecutionInfo.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 60, // 118: temporal.server.api.persistence.v1.ChildExecutionInfo.priority:type_name -> temporal.api.common.v1.Priority - 56, // 119: temporal.server.api.persistence.v1.RequestCancelInfo.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 56, // 120: temporal.server.api.persistence.v1.SignalInfo.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 81, // 121: temporal.server.api.persistence.v1.Checksum.flavor:type_name -> temporal.server.api.enums.v1.ChecksumFlavor - 42, // 122: temporal.server.api.persistence.v1.Callback.nexus:type_name -> temporal.server.api.persistence.v1.Callback.Nexus - 43, // 123: temporal.server.api.persistence.v1.Callback.hsm:type_name -> temporal.server.api.persistence.v1.Callback.HSM - 82, // 124: temporal.server.api.persistence.v1.Callback.links:type_name -> temporal.api.common.v1.Link - 83, // 125: temporal.server.api.persistence.v1.HSMCompletionCallbackArg.last_event:type_name -> temporal.api.history.v1.HistoryEvent - 23, // 126: temporal.server.api.persistence.v1.CallbackInfo.callback:type_name -> temporal.server.api.persistence.v1.Callback - 46, // 127: temporal.server.api.persistence.v1.CallbackInfo.trigger:type_name -> temporal.server.api.persistence.v1.CallbackInfo.Trigger - 47, // 128: temporal.server.api.persistence.v1.CallbackInfo.registration_time:type_name -> google.protobuf.Timestamp - 84, // 129: temporal.server.api.persistence.v1.CallbackInfo.state:type_name -> temporal.server.api.enums.v1.CallbackState - 47, // 130: temporal.server.api.persistence.v1.CallbackInfo.last_attempt_complete_time:type_name -> google.protobuf.Timestamp - 76, // 131: temporal.server.api.persistence.v1.CallbackInfo.last_attempt_failure:type_name -> temporal.api.failure.v1.Failure - 47, // 132: temporal.server.api.persistence.v1.CallbackInfo.next_attempt_schedule_time:type_name -> google.protobuf.Timestamp - 48, // 133: temporal.server.api.persistence.v1.NexusOperationInfo.schedule_to_close_timeout:type_name -> google.protobuf.Duration - 47, // 134: temporal.server.api.persistence.v1.NexusOperationInfo.scheduled_time:type_name -> google.protobuf.Timestamp - 85, // 135: temporal.server.api.persistence.v1.NexusOperationInfo.state:type_name -> temporal.server.api.enums.v1.NexusOperationState - 47, // 136: temporal.server.api.persistence.v1.NexusOperationInfo.last_attempt_complete_time:type_name -> google.protobuf.Timestamp - 76, // 137: temporal.server.api.persistence.v1.NexusOperationInfo.last_attempt_failure:type_name -> temporal.api.failure.v1.Failure - 47, // 138: temporal.server.api.persistence.v1.NexusOperationInfo.next_attempt_schedule_time:type_name -> google.protobuf.Timestamp - 48, // 139: temporal.server.api.persistence.v1.NexusOperationInfo.schedule_to_start_timeout:type_name -> google.protobuf.Duration - 48, // 140: temporal.server.api.persistence.v1.NexusOperationInfo.start_to_close_timeout:type_name -> google.protobuf.Duration - 47, // 141: temporal.server.api.persistence.v1.NexusOperationInfo.started_time:type_name -> google.protobuf.Timestamp - 47, // 142: temporal.server.api.persistence.v1.NexusOperationCancellationInfo.requested_time:type_name -> google.protobuf.Timestamp - 86, // 143: temporal.server.api.persistence.v1.NexusOperationCancellationInfo.state:type_name -> temporal.api.enums.v1.NexusOperationCancellationState - 47, // 144: temporal.server.api.persistence.v1.NexusOperationCancellationInfo.last_attempt_complete_time:type_name -> google.protobuf.Timestamp - 76, // 145: temporal.server.api.persistence.v1.NexusOperationCancellationInfo.last_attempt_failure:type_name -> temporal.api.failure.v1.Failure - 47, // 146: temporal.server.api.persistence.v1.NexusOperationCancellationInfo.next_attempt_schedule_time:type_name -> google.protobuf.Timestamp - 47, // 147: temporal.server.api.persistence.v1.WorkflowPauseInfo.pause_time:type_name -> google.protobuf.Timestamp - 87, // 148: temporal.server.api.persistence.v1.ShardInfo.QueueStatesEntry.value:type_name -> temporal.server.api.persistence.v1.QueueState - 88, // 149: temporal.server.api.persistence.v1.WorkflowExecutionInfo.SearchAttributesEntry.value:type_name -> temporal.api.common.v1.Payload - 88, // 150: temporal.server.api.persistence.v1.WorkflowExecutionInfo.MemoEntry.value:type_name -> temporal.api.common.v1.Payload - 89, // 151: temporal.server.api.persistence.v1.WorkflowExecutionInfo.UpdateInfosEntry.value:type_name -> temporal.server.api.persistence.v1.UpdateInfo - 90, // 152: temporal.server.api.persistence.v1.WorkflowExecutionInfo.SubStateMachinesByTypeEntry.value:type_name -> temporal.server.api.persistence.v1.StateMachineMap - 28, // 153: temporal.server.api.persistence.v1.WorkflowExecutionInfo.ChildrenInitializedPostResetPointEntry.value:type_name -> temporal.server.api.persistence.v1.ResetChildInfo - 7, // 154: temporal.server.api.persistence.v1.WorkflowExecutionState.RequestIdsEntry.value:type_name -> temporal.server.api.persistence.v1.RequestIDInfo - 47, // 155: temporal.server.api.persistence.v1.ActivityInfo.PauseInfo.pause_time:type_name -> google.protobuf.Timestamp - 41, // 156: temporal.server.api.persistence.v1.ActivityInfo.PauseInfo.manual:type_name -> temporal.server.api.persistence.v1.ActivityInfo.PauseInfo.Manual - 44, // 157: temporal.server.api.persistence.v1.Callback.Nexus.header:type_name -> temporal.server.api.persistence.v1.Callback.Nexus.HeaderEntry - 91, // 158: temporal.server.api.persistence.v1.Callback.HSM.ref:type_name -> temporal.server.api.persistence.v1.StateMachineRef - 45, // 159: temporal.server.api.persistence.v1.CallbackInfo.Trigger.workflow_closed:type_name -> temporal.server.api.persistence.v1.CallbackInfo.WorkflowClosed - 160, // [160:160] is the sub-list for method output_type - 160, // [160:160] is the sub-list for method input_type - 160, // [160:160] is the sub-list for extension type_name - 160, // [160:160] is the sub-list for extension extendee - 0, // [0:160] is the sub-list for field type_name + 47, // 61: temporal.server.api.persistence.v1.RequestIDInfo.attach_time:type_name -> google.protobuf.Timestamp + 69, // 62: temporal.server.api.persistence.v1.TransferTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType + 47, // 63: temporal.server.api.persistence.v1.TransferTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp + 38, // 64: temporal.server.api.persistence.v1.TransferTaskInfo.close_execution_task_details:type_name -> temporal.server.api.persistence.v1.TransferTaskInfo.CloseExecutionTaskDetails + 70, // 65: temporal.server.api.persistence.v1.TransferTaskInfo.chasm_task_info:type_name -> temporal.server.api.persistence.v1.ChasmTaskInfo + 69, // 66: temporal.server.api.persistence.v1.ReplicationTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType + 47, // 67: temporal.server.api.persistence.v1.ReplicationTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp + 71, // 68: temporal.server.api.persistence.v1.ReplicationTaskInfo.priority:type_name -> temporal.server.api.enums.v1.TaskPriority + 56, // 69: temporal.server.api.persistence.v1.ReplicationTaskInfo.versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 9, // 70: temporal.server.api.persistence.v1.ReplicationTaskInfo.task_equivalents:type_name -> temporal.server.api.persistence.v1.ReplicationTaskInfo + 72, // 71: temporal.server.api.persistence.v1.ReplicationTaskInfo.last_version_history_item:type_name -> temporal.server.api.history.v1.VersionHistoryItem + 69, // 72: temporal.server.api.persistence.v1.VisibilityTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType + 47, // 73: temporal.server.api.persistence.v1.VisibilityTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp + 47, // 74: temporal.server.api.persistence.v1.VisibilityTaskInfo.close_time:type_name -> google.protobuf.Timestamp + 47, // 75: temporal.server.api.persistence.v1.VisibilityTaskInfo.start_time:type_name -> google.protobuf.Timestamp + 70, // 76: temporal.server.api.persistence.v1.VisibilityTaskInfo.chasm_task_info:type_name -> temporal.server.api.persistence.v1.ChasmTaskInfo + 69, // 77: temporal.server.api.persistence.v1.TimerTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType + 62, // 78: temporal.server.api.persistence.v1.TimerTaskInfo.timeout_type:type_name -> temporal.api.enums.v1.TimeoutType + 73, // 79: temporal.server.api.persistence.v1.TimerTaskInfo.workflow_backoff_type:type_name -> temporal.server.api.enums.v1.WorkflowBackoffType + 47, // 80: temporal.server.api.persistence.v1.TimerTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp + 70, // 81: temporal.server.api.persistence.v1.TimerTaskInfo.chasm_task_info:type_name -> temporal.server.api.persistence.v1.ChasmTaskInfo + 56, // 82: temporal.server.api.persistence.v1.TimerTaskInfo.versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 69, // 83: temporal.server.api.persistence.v1.ArchivalTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType + 47, // 84: temporal.server.api.persistence.v1.ArchivalTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp + 69, // 85: temporal.server.api.persistence.v1.OutboundTaskInfo.task_type:type_name -> temporal.server.api.enums.v1.TaskType + 47, // 86: temporal.server.api.persistence.v1.OutboundTaskInfo.visibility_time:type_name -> google.protobuf.Timestamp + 74, // 87: temporal.server.api.persistence.v1.OutboundTaskInfo.state_machine_info:type_name -> temporal.server.api.persistence.v1.StateMachineTaskInfo + 70, // 88: temporal.server.api.persistence.v1.OutboundTaskInfo.chasm_task_info:type_name -> temporal.server.api.persistence.v1.ChasmTaskInfo + 14, // 89: temporal.server.api.persistence.v1.OutboundTaskInfo.worker_commands_task:type_name -> temporal.server.api.persistence.v1.WorkerCommandsTask + 75, // 90: temporal.server.api.persistence.v1.WorkerCommandsTask.commands:type_name -> temporal.api.worker.v1.WorkerCommand + 47, // 91: temporal.server.api.persistence.v1.ActivityInfo.scheduled_time:type_name -> google.protobuf.Timestamp + 47, // 92: temporal.server.api.persistence.v1.ActivityInfo.started_time:type_name -> google.protobuf.Timestamp + 48, // 93: temporal.server.api.persistence.v1.ActivityInfo.schedule_to_start_timeout:type_name -> google.protobuf.Duration + 48, // 94: temporal.server.api.persistence.v1.ActivityInfo.schedule_to_close_timeout:type_name -> google.protobuf.Duration + 48, // 95: temporal.server.api.persistence.v1.ActivityInfo.start_to_close_timeout:type_name -> google.protobuf.Duration + 48, // 96: temporal.server.api.persistence.v1.ActivityInfo.heartbeat_timeout:type_name -> google.protobuf.Duration + 48, // 97: temporal.server.api.persistence.v1.ActivityInfo.retry_initial_interval:type_name -> google.protobuf.Duration + 48, // 98: temporal.server.api.persistence.v1.ActivityInfo.retry_maximum_interval:type_name -> google.protobuf.Duration + 47, // 99: temporal.server.api.persistence.v1.ActivityInfo.retry_expiration_time:type_name -> google.protobuf.Timestamp + 76, // 100: temporal.server.api.persistence.v1.ActivityInfo.retry_last_failure:type_name -> temporal.api.failure.v1.Failure + 77, // 101: temporal.server.api.persistence.v1.ActivityInfo.last_heartbeat_details:type_name -> temporal.api.common.v1.Payloads + 47, // 102: temporal.server.api.persistence.v1.ActivityInfo.last_heartbeat_update_time:type_name -> google.protobuf.Timestamp + 78, // 103: temporal.server.api.persistence.v1.ActivityInfo.activity_type:type_name -> temporal.api.common.v1.ActivityType + 39, // 104: temporal.server.api.persistence.v1.ActivityInfo.use_workflow_build_id_info:type_name -> temporal.server.api.persistence.v1.ActivityInfo.UseWorkflowBuildIdInfo + 55, // 105: temporal.server.api.persistence.v1.ActivityInfo.last_worker_version_stamp:type_name -> temporal.api.common.v1.WorkerVersionStamp + 56, // 106: temporal.server.api.persistence.v1.ActivityInfo.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 47, // 107: temporal.server.api.persistence.v1.ActivityInfo.first_scheduled_time:type_name -> google.protobuf.Timestamp + 47, // 108: temporal.server.api.persistence.v1.ActivityInfo.last_attempt_complete_time:type_name -> google.protobuf.Timestamp + 79, // 109: temporal.server.api.persistence.v1.ActivityInfo.last_started_deployment:type_name -> temporal.api.deployment.v1.Deployment + 65, // 110: temporal.server.api.persistence.v1.ActivityInfo.last_deployment_version:type_name -> temporal.api.deployment.v1.WorkerDeploymentVersion + 60, // 111: temporal.server.api.persistence.v1.ActivityInfo.priority:type_name -> temporal.api.common.v1.Priority + 40, // 112: temporal.server.api.persistence.v1.ActivityInfo.pause_info:type_name -> temporal.server.api.persistence.v1.ActivityInfo.PauseInfo + 53, // 113: temporal.server.api.persistence.v1.ActivityInfo.started_clock:type_name -> temporal.server.api.clock.v1.VectorClock + 47, // 114: temporal.server.api.persistence.v1.TimerInfo.expiry_time:type_name -> google.protobuf.Timestamp + 56, // 115: temporal.server.api.persistence.v1.TimerInfo.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 80, // 116: temporal.server.api.persistence.v1.ChildExecutionInfo.parent_close_policy:type_name -> temporal.api.enums.v1.ParentClosePolicy + 53, // 117: temporal.server.api.persistence.v1.ChildExecutionInfo.clock:type_name -> temporal.server.api.clock.v1.VectorClock + 56, // 118: temporal.server.api.persistence.v1.ChildExecutionInfo.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 60, // 119: temporal.server.api.persistence.v1.ChildExecutionInfo.priority:type_name -> temporal.api.common.v1.Priority + 56, // 120: temporal.server.api.persistence.v1.RequestCancelInfo.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 56, // 121: temporal.server.api.persistence.v1.SignalInfo.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 81, // 122: temporal.server.api.persistence.v1.Checksum.flavor:type_name -> temporal.server.api.enums.v1.ChecksumFlavor + 42, // 123: temporal.server.api.persistence.v1.Callback.nexus:type_name -> temporal.server.api.persistence.v1.Callback.Nexus + 43, // 124: temporal.server.api.persistence.v1.Callback.hsm:type_name -> temporal.server.api.persistence.v1.Callback.HSM + 82, // 125: temporal.server.api.persistence.v1.Callback.links:type_name -> temporal.api.common.v1.Link + 83, // 126: temporal.server.api.persistence.v1.HSMCompletionCallbackArg.last_event:type_name -> temporal.api.history.v1.HistoryEvent + 23, // 127: temporal.server.api.persistence.v1.CallbackInfo.callback:type_name -> temporal.server.api.persistence.v1.Callback + 46, // 128: temporal.server.api.persistence.v1.CallbackInfo.trigger:type_name -> temporal.server.api.persistence.v1.CallbackInfo.Trigger + 47, // 129: temporal.server.api.persistence.v1.CallbackInfo.registration_time:type_name -> google.protobuf.Timestamp + 84, // 130: temporal.server.api.persistence.v1.CallbackInfo.state:type_name -> temporal.server.api.enums.v1.CallbackState + 47, // 131: temporal.server.api.persistence.v1.CallbackInfo.last_attempt_complete_time:type_name -> google.protobuf.Timestamp + 76, // 132: temporal.server.api.persistence.v1.CallbackInfo.last_attempt_failure:type_name -> temporal.api.failure.v1.Failure + 47, // 133: temporal.server.api.persistence.v1.CallbackInfo.next_attempt_schedule_time:type_name -> google.protobuf.Timestamp + 48, // 134: temporal.server.api.persistence.v1.NexusOperationInfo.schedule_to_close_timeout:type_name -> google.protobuf.Duration + 47, // 135: temporal.server.api.persistence.v1.NexusOperationInfo.scheduled_time:type_name -> google.protobuf.Timestamp + 85, // 136: temporal.server.api.persistence.v1.NexusOperationInfo.state:type_name -> temporal.server.api.enums.v1.NexusOperationState + 47, // 137: temporal.server.api.persistence.v1.NexusOperationInfo.last_attempt_complete_time:type_name -> google.protobuf.Timestamp + 76, // 138: temporal.server.api.persistence.v1.NexusOperationInfo.last_attempt_failure:type_name -> temporal.api.failure.v1.Failure + 47, // 139: temporal.server.api.persistence.v1.NexusOperationInfo.next_attempt_schedule_time:type_name -> google.protobuf.Timestamp + 48, // 140: temporal.server.api.persistence.v1.NexusOperationInfo.schedule_to_start_timeout:type_name -> google.protobuf.Duration + 48, // 141: temporal.server.api.persistence.v1.NexusOperationInfo.start_to_close_timeout:type_name -> google.protobuf.Duration + 47, // 142: temporal.server.api.persistence.v1.NexusOperationInfo.started_time:type_name -> google.protobuf.Timestamp + 47, // 143: temporal.server.api.persistence.v1.NexusOperationCancellationInfo.requested_time:type_name -> google.protobuf.Timestamp + 86, // 144: temporal.server.api.persistence.v1.NexusOperationCancellationInfo.state:type_name -> temporal.api.enums.v1.NexusOperationCancellationState + 47, // 145: temporal.server.api.persistence.v1.NexusOperationCancellationInfo.last_attempt_complete_time:type_name -> google.protobuf.Timestamp + 76, // 146: temporal.server.api.persistence.v1.NexusOperationCancellationInfo.last_attempt_failure:type_name -> temporal.api.failure.v1.Failure + 47, // 147: temporal.server.api.persistence.v1.NexusOperationCancellationInfo.next_attempt_schedule_time:type_name -> google.protobuf.Timestamp + 47, // 148: temporal.server.api.persistence.v1.WorkflowPauseInfo.pause_time:type_name -> google.protobuf.Timestamp + 87, // 149: temporal.server.api.persistence.v1.ShardInfo.QueueStatesEntry.value:type_name -> temporal.server.api.persistence.v1.QueueState + 88, // 150: temporal.server.api.persistence.v1.WorkflowExecutionInfo.SearchAttributesEntry.value:type_name -> temporal.api.common.v1.Payload + 88, // 151: temporal.server.api.persistence.v1.WorkflowExecutionInfo.MemoEntry.value:type_name -> temporal.api.common.v1.Payload + 89, // 152: temporal.server.api.persistence.v1.WorkflowExecutionInfo.UpdateInfosEntry.value:type_name -> temporal.server.api.persistence.v1.UpdateInfo + 90, // 153: temporal.server.api.persistence.v1.WorkflowExecutionInfo.SubStateMachinesByTypeEntry.value:type_name -> temporal.server.api.persistence.v1.StateMachineMap + 28, // 154: temporal.server.api.persistence.v1.WorkflowExecutionInfo.ChildrenInitializedPostResetPointEntry.value:type_name -> temporal.server.api.persistence.v1.ResetChildInfo + 7, // 155: temporal.server.api.persistence.v1.WorkflowExecutionState.RequestIdsEntry.value:type_name -> temporal.server.api.persistence.v1.RequestIDInfo + 47, // 156: temporal.server.api.persistence.v1.ActivityInfo.PauseInfo.pause_time:type_name -> google.protobuf.Timestamp + 41, // 157: temporal.server.api.persistence.v1.ActivityInfo.PauseInfo.manual:type_name -> temporal.server.api.persistence.v1.ActivityInfo.PauseInfo.Manual + 44, // 158: temporal.server.api.persistence.v1.Callback.Nexus.header:type_name -> temporal.server.api.persistence.v1.Callback.Nexus.HeaderEntry + 91, // 159: temporal.server.api.persistence.v1.Callback.HSM.ref:type_name -> temporal.server.api.persistence.v1.StateMachineRef + 45, // 160: temporal.server.api.persistence.v1.CallbackInfo.Trigger.workflow_closed:type_name -> temporal.server.api.persistence.v1.CallbackInfo.WorkflowClosed + 161, // [161:161] is the sub-list for method output_type + 161, // [161:161] is the sub-list for method input_type + 161, // [161:161] is the sub-list for extension type_name + 161, // [161:161] is the sub-list for extension extendee + 0, // [0:161] is the sub-list for field type_name } func init() { file_temporal_server_api_persistence_v1_executions_proto_init() } diff --git a/chasm/chasmtest/test_engine.go b/chasm/chasmtest/test_engine.go index cf88026f31..e4a10732fc 100644 --- a/chasm/chasmtest/test_engine.go +++ b/chasm/chasmtest/test_engine.go @@ -45,11 +45,12 @@ type ( } execution struct { - key chasm.ExecutionKey - node *chasm.Node - backend *chasm.MockNodeBackend - root chasm.RootComponent - requestID string + key chasm.ExecutionKey + node *chasm.Node + backend *chasm.MockNodeBackend + root chasm.RootComponent + createRequestID string + requestIDs map[string]struct{} // commitTransition advances the backend's committed transition count. commitTransition func() } @@ -137,8 +138,7 @@ func (e *Engine) StartExecution( current, hasCurrent := e.currentExecutions[bKey] if hasCurrent { - // if the requestID matches the original create request, return the existing run. - if options.RequestID != "" && options.RequestID == current.requestID { + if _, ok := current.requestIDs[options.RequestID]; options.RequestID != "" && ok { serializedRef, err := current.node.Ref(current.root) if err != nil { return chasm.StartExecutionResult{}, err @@ -179,7 +179,7 @@ func (e *Engine) UpdateWithStartExecution( if hasCurrent { switch current.backend.GetExecutionState().State { case enumsspb.WORKFLOW_EXECUTION_STATE_CREATED, enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING: - serializedRef, err := e.updateComponentInExecution(ctx, current, ref, updateFn) + serializedRef, err := e.updateComponentInExecution(ctx, current, ref, updateFn, "") if err != nil { return chasm.EngineUpdateWithStartExecutionResult{}, err } @@ -198,7 +198,7 @@ func (e *Engine) UpdateWithStartExecution( "CHASM execution already completed successfully. BusinessID: %s, RunID: %s, ID Reuse Policy: %v", ref.BusinessID, current.key.RunID, options.ReusePolicy, ), - current.requestID, + current.createRequestID, current.key.RunID, ) } @@ -208,7 +208,7 @@ func (e *Engine) UpdateWithStartExecution( "CHASM execution already finished. BusinessID: %s, RunID: %s, ID Reuse Policy: %v", ref.BusinessID, current.key.RunID, options.ReusePolicy, ), - current.requestID, + current.createRequestID, current.key.RunID, ) default: @@ -230,13 +230,22 @@ func (e *Engine) UpdateComponent( ctx context.Context, ref chasm.ComponentRef, updateFn func(chasm.MutableContext, chasm.Component) error, - _ ...chasm.TransitionOption, + opts ...chasm.TransitionOption, ) ([]byte, error) { execution, err := e.executionForRef(ref) if err != nil { return nil, err } - return e.updateComponentInExecution(ctx, execution, ref, updateFn) + + options := constructTransitionOptions(opts...) + if options.RequestID != "" { + if _, ok := execution.requestIDs[options.RequestID]; ok { + return nil, serviceerror.NewFailedPreconditionf( + "request ID %s has already been used for this execution", options.RequestID) + } + } + + return e.updateComponentInExecution(ctx, execution, ref, updateFn, options.RequestID) } func (e *Engine) ReadComponent( @@ -360,7 +369,7 @@ func (e *Engine) handleConflictPolicy( "CHASM execution still running. BusinessID: %s, RunID: %s, ID Conflict Policy: %v", ref.BusinessID, current.key.RunID, options.ConflictPolicy, ), - current.requestID, + current.createRequestID, current.key.RunID, ) case chasm.BusinessIDConflictPolicyTerminateExisting: @@ -404,7 +413,7 @@ func (e *Engine) handleReusePolicy( "CHASM execution already completed successfully. BusinessID: %s, RunID: %s, ID Reuse Policy: %v", ref.BusinessID, current.key.RunID, options.ReusePolicy, ), - current.requestID, + current.createRequestID, current.key.RunID, ) } @@ -414,7 +423,7 @@ func (e *Engine) handleReusePolicy( "CHASM execution already finished. BusinessID: %s, RunID: %s, ID Reuse Policy: %v", ref.BusinessID, current.key.RunID, options.ReusePolicy, ), - current.requestID, + current.createRequestID, current.key.RunID, ) default: @@ -433,7 +442,8 @@ func (e *Engine) startNew( requestID string, ) (chasm.StartExecutionResult, error) { exec := e.newExecution(key) - exec.requestID = requestID + exec.createRequestID = requestID + exec.recordRequestID(requestID) mutableCtx := chasm.NewMutableContext(ctx, exec.node) root, err := startFn(mutableCtx) @@ -473,7 +483,8 @@ func (e *Engine) startAndUpdateNew( requestID string, ) (chasm.EngineUpdateWithStartExecutionResult, error) { exec := e.newExecution(key) - exec.requestID = requestID + exec.createRequestID = requestID + exec.recordRequestID(requestID) mutableCtx := chasm.NewMutableContext(ctx, exec.node) root, err := startFn(mutableCtx) @@ -595,6 +606,16 @@ func (x *execution) closeTransaction() error { return nil } +func (x *execution) recordRequestID(requestID string) { + if requestID == "" { + return + } + if x.requestIDs == nil { + x.requestIDs = make(map[string]struct{}) + } + x.requestIDs[requestID] = struct{}{} +} + // executionForRef looks up an execution by the ref's RunID when present, or falls back // to the current run for the business ID when RunID is empty. func (e *Engine) executionForRef(ref chasm.ComponentRef) (*execution, error) { @@ -621,6 +642,7 @@ func (e *Engine) updateComponentInExecution( execution *execution, ref chasm.ComponentRef, updateFn func(chasm.MutableContext, chasm.Component) error, + requestID string, ) ([]byte, error) { mutableCtx := chasm.NewMutableContext(ctx, execution.node) component, err := execution.node.Component(mutableCtx, ref) @@ -635,8 +657,13 @@ func (e *Engine) updateComponentInExecution( if err = execution.closeTransaction(); err != nil { return nil, err } + execution.recordRequestID(requestID) - return mutableCtx.Ref(component) + serializedRef, err := mutableCtx.Ref(component) + if errors.As(err, new(*serviceerror.NotFound)) { + return nil, nil + } + return serializedRef, err } // refForComponent looks up the ComponentRef for a component instance by scanning diff --git a/chasm/chasmtest/test_engine_test.go b/chasm/chasmtest/test_engine_test.go index 55afe7e1cb..ec40d7f171 100644 --- a/chasm/chasmtest/test_engine_test.go +++ b/chasm/chasmtest/test_engine_test.go @@ -3,10 +3,12 @@ package chasmtest_test import ( "context" + "errors" "testing" "time" "github.com/stretchr/testify/require" + "go.temporal.io/api/serviceerror" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/chasmtest" "go.temporal.io/server/chasm/lib/tests" @@ -39,7 +41,88 @@ func TestTasksArePhysicallyGenerated(t *testing.T) { }) } -func startStore(t *testing.T, ttl time.Duration) (*chasmtest.Engine, chasm.ComponentRef) { +func TestUpdateComponentDeduplicatesRequestID(t *testing.T) { + e, ref := startStore(t, 0) + updateCount := 0 + update := func() ([]byte, error) { + _, updatedRef, err := chasm.UpdateComponent(engineContext(e), ref, + func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) { + updateCount++ + return nil, nil + }, nil, chasm.WithRequestID("request-id")) + return updatedRef, err + } + + updatedRef, err := update() + require.NoError(t, err) + require.NotEmpty(t, updatedRef) + + updatedRef, err = update() + var failedPrecondition *serviceerror.FailedPrecondition + require.ErrorAs(t, err, &failedPrecondition) + require.Nil(t, updatedRef) + require.Equal(t, 1, updateCount) +} + +func TestUpdateComponentDoesNotRecordFailedRequestID(t *testing.T) { + e, ref := startStore(t, 0) + updateErr := errors.New("update failed") + updateCount := 0 + update := func(errToReturn error) error { + _, _, err := chasm.UpdateComponent(engineContext(e), ref, + func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) { + updateCount++ + return nil, errToReturn + }, nil, chasm.WithRequestID("request-id")) + return err + } + + require.ErrorIs(t, update(updateErr), updateErr) + require.NoError(t, update(nil)) + require.Equal(t, 2, updateCount) +} + +func TestUpdateComponentDeduplicatesCreationRequestID(t *testing.T) { + e, ref := startStore(t, 0, chasm.WithRequestID("request-id")) + updateCount := 0 + _, _, err := chasm.UpdateComponent(engineContext(e), ref, + func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) { + updateCount++ + return nil, nil + }, nil, chasm.WithRequestID("request-id")) + var failedPrecondition *serviceerror.FailedPrecondition + require.ErrorAs(t, err, &failedPrecondition) + require.Equal(t, 0, updateCount) +} + +func TestStartExecutionDeduplicatesUpdateRequestID(t *testing.T) { + e, ref := startStore(t, 0) + ctx := engineContext(e) + _, _, err := chasm.UpdateComponent(ctx, ref, + func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) { + return nil, nil + }, nil, chasm.WithRequestID("request-id")) + require.NoError(t, err) + + startCount := 0 + startKey := ref.ExecutionKey + startKey.RunID = "" + result, err := chasm.StartExecution(ctx, startKey, + func(chasm.MutableContext, any) (*tests.PayloadStore, error) { + startCount++ + return nil, nil + }, nil, chasm.WithRequestID("request-id")) + require.NoError(t, err) + require.False(t, result.Created) + require.Equal(t, ref.ExecutionKey, result.ExecutionKey) + require.Equal(t, 0, startCount) +} + +func startStore( + t *testing.T, + ttl time.Duration, + opts ...chasm.TransitionOption, +) (*chasmtest.Engine, chasm.ComponentRef) { registry := chasm.NewRegistry(log.NewNoopLogger()) require.NoError(t, registry.Register(&chasm.CoreLibrary{})) require.NoError(t, registry.Register(tests.Library)) @@ -56,7 +139,7 @@ func startStore(t *testing.T, ttl time.Duration) (*chasmtest.Engine, chasm.Compo return nil, err } return store, addPayload(store, mc, "first", ttl) - }, nil) + }, nil, opts...) require.NoError(t, err) key.RunID = result.ExecutionKey.RunID diff --git a/chasm/engine.go b/chasm/engine.go index ea92467d5a..6959b6aff7 100644 --- a/chasm/engine.go +++ b/chasm/engine.go @@ -190,8 +190,12 @@ func WithBusinessIDPolicy( } } -// WithRequestID sets the requestID used when creating a new execution. -// This option only applies to StartExecution() and UpdateWithStartExecution(). +// WithRequestID sets the requestID for the transition. +// +// On StartExecution() and UpdateWithStartExecution() it is the request ID +// recorded when creating a new execution. On UpdateComponent(), it is used for +// execution-level idempotency, recording a request's ID on success, and failing +// subsequent requests reusing a request ID by returning a FailedPrecondition error. func WithRequestID( requestID string, ) TransitionOption { @@ -326,12 +330,14 @@ func UpdateWithStartExecution[C RootComponent, I any, O any]( // // UpdateComponent applies updateFn to the component identified by the supplied component reference. // -// The only opts currently honored is [WithRefConsistencyLevel]; it selects the [RefConsistencyLevel] used to -// resolve and validate the ref (see that type for the ladder of levels). Other options are ignored. +// Two opts are honored: [WithRefConsistencyLevel] selects the [RefConsistencyLevel] used to resolve +// and validate the ref (see that type for the ladder of levels); [WithRequestID] enables +// execution-level idempotency, rejecting a repeated update that carries an already-recorded request +// ID (see that option). Other options are ignored. // // It returns the result, along with the new component reference. The returned reference may be // nil when updateFn deletes the component in the same transaction and the component is not the -// root component. +// root component, or on the [WithRequestID] dedup rejection (along with a FailedPrecondition error). func UpdateComponent[C any, R []byte | ComponentRef, I any, O any]( ctx context.Context, r R, diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go index 1e3bf1f624..a046b85b0f 100644 --- a/common/dynamicconfig/constants.go +++ b/common/dynamicconfig/constants.go @@ -2608,6 +2608,18 @@ system.transactionSizeLimit, since each batch is persisted within a single trans 10000, `MaximumSignalsPerExecution is max number of signals supported by single execution`, ) + MaximumRequestIDsPerExecution = NewNamespaceIntSetting( + "history.maximumRequestIDsPerExecution", + 25, + `MaximumRequestIDsPerExecution is the hard cap on CHASM-attached request IDs retained per +execution for UpdateComponent idempotency; the oldest are swept beyond this limit. Set to 0 to disable the count cap.`, + ) + RequestIDMaxAge = NewNamespaceDurationSetting( + "history.requestIDMaxAge", + 7*24*time.Hour, + `RequestIDMaxAge is the maximum age of a CHASM-attached request ID retained per execution for +UpdateComponent idempotency. Set to 0 to disable age-based sweeping.`, + ) ShardUpdateMinInterval = NewGlobalDurationSetting( "history.shardUpdateMinInterval", 5*time.Minute, diff --git a/common/metrics/metric_defs.go b/common/metrics/metric_defs.go index 81d465dffd..6b43228f18 100644 --- a/common/metrics/metric_defs.go +++ b/common/metrics/metric_defs.go @@ -938,6 +938,10 @@ var ( "chasm_incoming_signal_duplicate", WithDescription("The number of duplicate signal request IDs detected when writing to the CHASM IncomingSignals map. Non-zero values indicate unexpected signal redelivery."), ) + CHASMRequestIDEvicted = NewCounterDef( + "chasm_request_id_evicted", + WithDescription("The number of CHASM-attached request IDs swept from an execution's dedup map for exceeding history.maximumRequestIDsPerExecution or history.requestIDMaxAge."), + ) TaskScheduleToStartLatency = NewTimerDef("task_schedule_to_start_latency") TaskBatchCompleteCounter = NewCounterDef("task_batch_complete_counter") TaskReschedulerPendingTasks = NewDimensionlessHistogramDef("task_rescheduler_pending_tasks") diff --git a/proto/internal/temporal/server/api/persistence/v1/executions.proto b/proto/internal/temporal/server/api/persistence/v1/executions.proto index 4d2be0f15a..f28631a455 100644 --- a/proto/internal/temporal/server/api/persistence/v1/executions.proto +++ b/proto/internal/temporal/server/api/persistence/v1/executions.proto @@ -386,8 +386,11 @@ message WorkflowExecutionState { VersionedTransition last_update_versioned_transition = 5; google.protobuf.Timestamp start_time = 6; // Request IDs that are attached to the workflow execution. It can be the request ID that started - // the workflow execution or request IDs that were attached to an existing running workflow - // execution via StartWorkflowExecutionRequest.OnConflictOptions. + // the workflow execution, request IDs that were attached to an existing running workflow + // execution via StartWorkflowExecutionRequest.OnConflictOptions, or (for CHASM executions) request + // IDs recorded for UpdateComponent idempotency - the latter are marked by RequestIDInfo.attach_time + // and are swept per-execution by count (history.maximumRequestIDsPerExecution) and age + // (history.requestIDMaxAge). map request_ids = 7; // Run ID of the first execution in the chain (set on WorkflowExecutionStarted). Equals run_id // only for a first run; any continuation (continue-as-new, retry, cron, or reset) @@ -399,6 +402,9 @@ message WorkflowExecutionState { message RequestIDInfo { temporal.api.enums.v1.EventType event_type = 1; int64 event_id = 2; + // Set only for request IDs attached by the CHASM framework for UpdateComponent idempotency. + // Used as an ordering key for lazy eviction when set. + google.protobuf.Timestamp attach_time = 3; } // transfer column diff --git a/service/history/chasm_engine.go b/service/history/chasm_engine.go index e86cf2480e..f9fde9abdb 100644 --- a/service/history/chasm_engine.go +++ b/service/history/chasm_engine.go @@ -370,7 +370,9 @@ func (e *ChasmEngine) updateExecution( actualRef := executionRef actualRef.RunID = workflowKey.RunID - serializedRef, err := e.applyUpdateWithLease(ctx, shardContext, executionLease, actualRef, updateFn) + // TODO(awln-temporal): Add a separate execution-scoped request-ID set for + // UpdateWithStart updates. Retries currently reapply updateFn. + serializedRef, err := e.applyUpdateWithLease(ctx, shardContext, executionLease, actualRef, updateFn, "") if err != nil { return chasm.ExecutionKey{}, nil, err } @@ -417,6 +419,7 @@ func (e *ChasmEngine) applyUpdateWithLease( executionLease api.WorkflowLease, ref chasm.ComponentRef, updateFn func(chasm.MutableContext, chasm.Component) error, + requestID string, ) ([]byte, error) { mutableState := executionLease.GetMutableState() chasmTree, err := chasmTreeFromMutableState(shardContext.GetLogger(), mutableState) @@ -434,6 +437,12 @@ func (e *ChasmEngine) applyUpdateWithLease( return nil, err } + // Record the accepted request ID so a later reuse is deduplicated (see AttachChasmRequestID). + // The dedup check itself runs earlier in updateComponent, before the lease work. + if requestID != "" { + mutableState.AttachChasmRequestID(requestID) + } + // TODO: Support WithSpeculative() TransitionOption. e.setContextMetadata(ctx, chasmTree) @@ -516,7 +525,10 @@ func (e *ChasmEngine) startAndUpdateExecution( // UpdateComponent applies updateFn to the component identified by the supplied component reference, // returning the new component reference corresponding to the transition. An error is returned if // the state transition specified by the supplied component reference is inconsistent with execution -// transition history. opts are currently ignored. +// transition history. +// +// A request ID supplied via chasm.WithRequestID deduplicates the update (see that option); other +// opts are ignored on this path. func (e *ChasmEngine) UpdateComponent( ctx context.Context, ref chasm.ComponentRef, @@ -524,7 +536,18 @@ func (e *ChasmEngine) UpdateComponent( opts ...chasm.TransitionOption, ) ([]byte, error) { options := e.constructTransitionOptions(opts...) - result, err := e.updateComponent(ctx, ref, updateFn) + + // constructTransitionOptions generates a RequestID for the benefit of + // traceability through logs, when none was supplied as part of the API request. + // Because this server-side RequestID isn't useful for idempotency, it isn't + // recorded as part of updateComponent, and so we only pass RequestID for recording + // when it comes from the client side. + var explicit chasm.TransitionOptions + for _, opt := range opts { + opt(&explicit) + } + + result, err := e.updateComponent(ctx, ref, updateFn, explicit.RequestID) return result, e.convertError(err, ref, options.RequestID) } @@ -532,17 +555,26 @@ func (e *ChasmEngine) updateComponent( ctx context.Context, ref chasm.ComponentRef, updateFn func(chasm.MutableContext, chasm.Component) error, + requestID string, ) (updatedRef []byte, retError error) { shardContext, executionLease, err := e.getExecutionLease(ctx, ref) if err != nil { return nil, err } + // When a request ID has already been recorded, we know this is a retry of a successful call, so + // reject it as non-retryable (FailedPrecondition) without running updateFn or persisting. + if requestID != "" && executionLease.GetMutableState().HasRequestID(requestID) { + executionLease.GetReleaseFn()(nil) + return nil, serviceerror.NewFailedPreconditionf( + "request ID %s has already been used for this execution", requestID) + } + defer func() { executionLease.GetReleaseFn()(retError) }() - return e.applyUpdateWithLease(ctx, shardContext, executionLease, ref, updateFn) + return e.applyUpdateWithLease(ctx, shardContext, executionLease, ref, updateFn, requestID) } // DeleteExecution deletes a CHASM execution. If the execution is still running on the active diff --git a/service/history/chasm_engine_test.go b/service/history/chasm_engine_test.go index 71d61b0b6a..9eb12ccadb 100644 --- a/service/history/chasm_engine_test.go +++ b/service/history/chasm_engine_test.go @@ -925,6 +925,230 @@ func (s *chasmEngineSuite) TestUpdateComponent_SetsContextMetadata() { s.assertTestContextMetadata(requestCtx, newActivityID, "update-request") } +func (s *chasmEngineSuite) TestUpdateComponent_RequestIDIdempotency() { + tv := testvars.New(s.T()) + tv = tv.WithRunID(tv.Any().RunID()) + + ref := chasm.NewComponentRef[*testComponent]( + chasm.ExecutionKey{ + NamespaceID: string(tests.NamespaceID), + BusinessID: tv.WorkflowID(), + RunID: tv.RunID(), + }, + ) + + requestID := tv.RequestID() + + s.mockExecutionManager.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()). + Return(&persistence.GetWorkflowExecutionResponse{ + State: s.buildPersistenceMutableState(ref.ExecutionKey, &persistencespb.ActivityInfo{ + ActivityId: "", + }, enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, nil), + }, nil).Times(1) // the duplicate update reads the cached, post-first-update state + // Only the first (non-duplicate) update persists and notifies; it must durably record the ID. + s.mockExecutionManager.EXPECT().UpdateWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, request *persistence.UpdateWorkflowExecutionRequest) (*persistence.UpdateWorkflowExecutionResponse, error) { + s.Contains(request.UpdateWorkflowMutation.ExecutionState.RequestIds, requestID, + "the accepted request ID must be persisted for durable dedup") + return tests.UpdateWorkflowExecutionResponse, nil + }).Times(1) + s.mockEngine.EXPECT().NotifyChasmExecution(ref.ExecutionKey, gomock.Any()).Return().Times(1) + + updateCount := 0 + update := func() ([]byte, error) { + return s.engine.UpdateComponent( + context.Background(), + ref, + func(ctx chasm.MutableContext, component chasm.Component) error { + updateCount++ + tc, ok := component.(*testComponent) + s.True(ok) + tc.ActivityInfo.ActivityId = tv.ActivityID() + return nil + }, + chasm.WithRequestID(requestID), + ) + } + + // First call runs updateFn and records the request ID. + firstRef, err := update() + s.NoError(err) + s.NotEmpty(firstRef) + s.Equal(1, updateCount) + + // Second call with the same request ID is rejected with a FailedPrecondition error: updateFn does + // not run and nothing is persisted. + dupRef, err := update() + var failedPrecondition *serviceerror.FailedPrecondition + s.ErrorAs(err, &failedPrecondition) + s.Nil(dupRef) + s.Equal(1, updateCount, "updateFn must not run again for a duplicate request ID") +} + +// TestUpdateComponent_NoRequestID verifies that without WithRequestID (or with an empty request ID) +// updates are not deduplicated - updateFn runs every time and no request ID is recorded. This guards +// the deliberate choice not to auto-generate a request ID on the update path. +func (s *chasmEngineSuite) TestUpdateComponent_NoRequestID() { + testCases := []struct { + name string + opts []chasm.TransitionOption + }{ + {"no option", nil}, + {"empty request ID", []chasm.TransitionOption{chasm.WithRequestID("")}}, + } + for _, tc := range testCases { + s.Run(tc.name, func() { + tv := testvars.New(s.T()) + tv = tv.WithRunID(tv.Any().RunID()) + ref := chasm.NewComponentRef[*testComponent](chasm.ExecutionKey{ + NamespaceID: string(tests.NamespaceID), + BusinessID: tv.WorkflowID(), + RunID: tv.RunID(), + }) + + s.mockExecutionManager.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()). + Return(&persistence.GetWorkflowExecutionResponse{ + State: s.buildPersistenceMutableState(ref.ExecutionKey, &persistencespb.ActivityInfo{}, + enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, nil), + }, nil).Times(1) + s.mockExecutionManager.EXPECT().UpdateWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, request *persistence.UpdateWorkflowExecutionRequest) (*persistence.UpdateWorkflowExecutionResponse, error) { + s.Empty(request.UpdateWorkflowMutation.ExecutionState.RequestIds, + "no request ID should be recorded when none is supplied") + return tests.UpdateWorkflowExecutionResponse, nil + }).Times(2) + s.mockEngine.EXPECT().NotifyChasmExecution(ref.ExecutionKey, gomock.Any()).Return().Times(2) + + updateCount := 0 + for range 2 { + _, err := s.engine.UpdateComponent( + context.Background(), + ref, + func(ctx chasm.MutableContext, component chasm.Component) error { + updateCount++ + component.(*testComponent).ActivityInfo.ActivityId = fmt.Sprintf("act-%d", updateCount) + return nil + }, + tc.opts..., + ) + s.NoError(err) + } + s.Equal(2, updateCount, "updateFn must run on every call when not deduplicated") + }) + } +} + +// TestUpdateComponent_DifferentRequestIDsWithEviction verifies that distinct request IDs each run and +// are recorded, and that the lazy sweep fires end-to-end through the transaction: with a limit of 1, +// the older request ID is evicted from the persisted map when the second is recorded. +func (s *chasmEngineSuite) TestUpdateComponent_DifferentRequestIDsWithEviction() { + tv := testvars.New(s.T()) + tv = tv.WithRunID(tv.Any().RunID()) + ref := chasm.NewComponentRef[*testComponent](chasm.ExecutionKey{ + NamespaceID: string(tests.NamespaceID), + BusinessID: tv.WorkflowID(), + RunID: tv.RunID(), + }) + s.config.MaximumRequestIDsPerExecution = func(string) int { return 1 } + requestID1 := "request-id-1" + requestID2 := "request-id-2" + + s.mockExecutionManager.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()). + Return(&persistence.GetWorkflowExecutionResponse{ + State: s.buildPersistenceMutableState(ref.ExecutionKey, &persistencespb.ActivityInfo{}, + enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, nil), + }, nil).Times(1) + var persisted []map[string]struct{} + s.mockExecutionManager.EXPECT().UpdateWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, request *persistence.UpdateWorkflowExecutionRequest) (*persistence.UpdateWorkflowExecutionResponse, error) { + snap := map[string]struct{}{} + for id := range request.UpdateWorkflowMutation.ExecutionState.RequestIds { + snap[id] = struct{}{} + } + persisted = append(persisted, snap) + return tests.UpdateWorkflowExecutionResponse, nil + }).Times(2) + s.mockEngine.EXPECT().NotifyChasmExecution(ref.ExecutionKey, gomock.Any()).Return().Times(2) + + updateCount := 0 + for _, id := range []string{requestID1, requestID2} { + _, err := s.engine.UpdateComponent( + context.Background(), + ref, + func(ctx chasm.MutableContext, component chasm.Component) error { + updateCount++ + component.(*testComponent).ActivityInfo.ActivityId = fmt.Sprintf("act-%d", updateCount) + return nil + }, + chasm.WithRequestID(id), + ) + s.NoError(err) + } + + s.Equal(2, updateCount, "distinct request IDs must each run updateFn") + s.Contains(persisted[0], requestID1) + // With a limit of 1, recording the second request ID sweeps the first. + s.Contains(persisted[1], requestID2) + s.NotContains(persisted[1], requestID1, "the older request ID must be swept when over the limit") +} + +// TestUpdateComponent_FailedUpdateNotRecorded verifies that when updateFn returns an error the request +// ID is not recorded, so a retry with the same request ID is not deduplicated and runs again. +func (s *chasmEngineSuite) TestUpdateComponent_FailedUpdateNotRecorded() { + tv := testvars.New(s.T()) + tv = tv.WithRunID(tv.Any().RunID()) + ref := chasm.NewComponentRef[*testComponent](chasm.ExecutionKey{ + NamespaceID: string(tests.NamespaceID), + BusinessID: tv.WorkflowID(), + RunID: tv.RunID(), + }) + requestID := tv.RequestID() + + // The failed update releases the lease with an error, clearing the cache, so the retry reloads. + s.mockExecutionManager.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn( + func(context.Context, *persistence.GetWorkflowExecutionRequest) (*persistence.GetWorkflowExecutionResponse, error) { + return &persistence.GetWorkflowExecutionResponse{ + State: s.buildPersistenceMutableState(ref.ExecutionKey, &persistencespb.ActivityInfo{}, + enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, nil), + }, nil + }).Times(2) + // Only the successful retry persists, and it records the request ID. + s.mockExecutionManager.EXPECT().UpdateWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, request *persistence.UpdateWorkflowExecutionRequest) (*persistence.UpdateWorkflowExecutionResponse, error) { + s.Contains(request.UpdateWorkflowMutation.ExecutionState.RequestIds, requestID) + return tests.UpdateWorkflowExecutionResponse, nil + }).Times(1) + s.mockEngine.EXPECT().NotifyChasmExecution(ref.ExecutionKey, gomock.Any()).Return().Times(1) + + updateErr := errors.New("update failed") + updateCount := 0 + update := func(fail bool) error { + _, err := s.engine.UpdateComponent( + context.Background(), + ref, + func(ctx chasm.MutableContext, component chasm.Component) error { + updateCount++ + if fail { + return updateErr + } + component.(*testComponent).ActivityInfo.ActivityId = tv.ActivityID() + return nil + }, + chasm.WithRequestID(requestID), + ) + return err + } + + // The update fails inside updateFn: nothing is persisted and the request ID is not recorded. + s.ErrorIs(update(true), updateErr) + s.Equal(1, updateCount) + + // Retrying with the same request ID is not deduplicated (the failure recorded nothing), so + // updateFn runs again and the update succeeds. + s.NoError(update(false)) + s.Equal(2, updateCount) +} + func (s *chasmEngineSuite) TestReadComponent_Success() { tv := testvars.New(s.T()) tv = tv.WithRunID(tv.Any().RunID()) diff --git a/service/history/configs/config.go b/service/history/configs/config.go index 0f019e44ea..44c5e6ba81 100644 --- a/service/history/configs/config.go +++ b/service/history/configs/config.go @@ -208,6 +208,8 @@ type Config struct { MaximumBufferedEventsBatch dynamicconfig.IntPropertyFn MaximumBufferedEventsSizeInBytes dynamicconfig.IntPropertyFn MaximumSignalsPerExecution dynamicconfig.IntPropertyFnWithNamespaceFilter + MaximumRequestIDsPerExecution dynamicconfig.IntPropertyFnWithNamespaceFilter + RequestIDMaxAge dynamicconfig.DurationPropertyFnWithNamespaceFilter MaximumEventBatchSizeInBytes dynamicconfig.IntPropertyFn // ShardUpdateMinInterval is the minimum time interval within which the shard info can be updated. @@ -667,6 +669,8 @@ func NewConfig( MaximumBufferedEventsBatch: dynamicconfig.MaximumBufferedEventsBatch.Get(dc), MaximumBufferedEventsSizeInBytes: dynamicconfig.MaximumBufferedEventsSizeInBytes.Get(dc), MaximumSignalsPerExecution: dynamicconfig.MaximumSignalsPerExecution.Get(dc), + MaximumRequestIDsPerExecution: dynamicconfig.MaximumRequestIDsPerExecution.Get(dc), + RequestIDMaxAge: dynamicconfig.RequestIDMaxAge.Get(dc), MaximumEventBatchSizeInBytes: dynamicconfig.MaximumEventBatchSizeInBytes.Get(dc), ShardUpdateMinInterval: dynamicconfig.ShardUpdateMinInterval.Get(dc), ShardFirstUpdateInterval: dynamicconfig.ShardFirstUpdateInterval.Get(dc), diff --git a/service/history/interfaces/mutable_state.go b/service/history/interfaces/mutable_state.go index b62c664be1..dea058064a 100644 --- a/service/history/interfaces/mutable_state.go +++ b/service/history/interfaces/mutable_state.go @@ -148,6 +148,9 @@ type ( SetChildrenInitializedPostResetPoint(children map[string]*persistencespb.ResetChildInfo) GetChildrenInitializedPostResetPoint() map[string]*persistencespb.ResetChildInfo AttachRequestID(requestID string, eventType enumspb.EventType, eventID int64) + // AttachChasmRequestID records a request ID attached by the CHASM framework for + // UpdateComponent idempotency. Such entries are sweepable (oldest-first) at transaction close. + AttachChasmRequestID(requestID string) CloneToProto() *persistencespb.WorkflowMutableState RetryActivity(ai *persistencespb.ActivityInfo, failure *failurepb.Failure) (enumspb.RetryState, error) diff --git a/service/history/interfaces/mutable_state_mock.go b/service/history/interfaces/mutable_state_mock.go index 599eaf8abd..87415d5a0b 100644 --- a/service/history/interfaces/mutable_state_mock.go +++ b/service/history/interfaces/mutable_state_mock.go @@ -1696,6 +1696,18 @@ func (mr *MockMutableStateMockRecorder) ApplyWorkflowTaskTimedOutEvent(arg0 any) return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ApplyWorkflowTaskTimedOutEvent", reflect.TypeOf((*MockMutableState)(nil).ApplyWorkflowTaskTimedOutEvent), arg0) } +// AttachChasmRequestID mocks base method. +func (m *MockMutableState) AttachChasmRequestID(requestID string) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "AttachChasmRequestID", requestID) +} + +// AttachChasmRequestID indicates an expected call of AttachChasmRequestID. +func (mr *MockMutableStateMockRecorder) AttachChasmRequestID(requestID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AttachChasmRequestID", reflect.TypeOf((*MockMutableState)(nil).AttachChasmRequestID), requestID) +} + // AttachRequestID mocks base method. func (m *MockMutableState) AttachRequestID(requestID string, eventType enums.EventType, eventID int64) { m.ctrl.T.Helper() diff --git a/service/history/workflow/mutable_state_impl.go b/service/history/workflow/mutable_state_impl.go index e8c9f474f6..828202c5a0 100644 --- a/service/history/workflow/mutable_state_impl.go +++ b/service/history/workflow/mutable_state_impl.go @@ -203,7 +203,11 @@ type ( visibilityUpdated bool executionStateUpdated bool workflowTaskUpdated bool - updateInfoUpdated map[string]struct{} + // chasmRequestIDsAdded is set when a CHASM request ID is attached in the current transaction + // (via AttachChasmRequestID). It gates the lazy request-ID sweep at transaction close so only + // transactions that added an ID pay for the scan. + chasmRequestIDsAdded bool + updateInfoUpdated map[string]struct{} // following xxxUserDataUpdated fields are for tracking if activity/timer user data updated. // This help to determine if we need to update transition history: For // user data change, we need to update transition history. No update for @@ -2600,6 +2604,41 @@ func (ms *MutableStateImpl) AttachRequestID( ms.approximateSize += ms.executionState.Size() } +// AttachChasmRequestID records a request ID attached by the CHASM framework for UpdateComponent +// idempotency. Unlike AttachRequestID (used for the create request ID and event-backed request IDs), +// the entry carries a non-nil AttachTime, which marks it as CHASM-attached (and therefore sweepable) +// and serves as the oldest-first ordering key for the lazy sweep at transaction close +// (see closeTransactionSweepChasmRequestIDs). It never touches CreateRequestId. +func (ms *MutableStateImpl) AttachChasmRequestID(requestID string) { + if requestID == "" { + return + } + // The caller (updateComponent) dedups via HasRequestID first, so an already-recorded ID here is a + // bug: the write below is unconditional and would overwrite the existing entry. + softassert.That(ms.logger, !ms.HasRequestID(requestID), + "AttachChasmRequestID called with an already-recorded request ID") + if ms.executionState.RequestIds == nil { + ms.executionState.RequestIds = make(map[string]*persistencespb.RequestIDInfo, 1) + } + entry := &persistencespb.RequestIDInfo{ + EventType: enumspb.EVENT_TYPE_UNSPECIFIED, + EventId: common.EmptyEventID, + AttachTime: timestamppb.New(ms.timeSource.Now()), + } + ms.executionState.RequestIds[requestID] = entry + ms.approximateSize += chasmRequestIDApproxSize(requestID, entry) + // Force the execution state blob to be persisted even if the CHASM tree made no other change, + // so the recorded request ID is durable. Also gate the lazy sweep on this transaction. + ms.executionStateUpdated = true + ms.chasmRequestIDsAdded = true +} + +// chasmRequestIDApproxSize approximates a CHASM request-ID map entry's contribution to +// approximateSize. +func chasmRequestIDApproxSize(requestID string, info *persistencespb.RequestIDInfo) int { + return len(requestID) + info.Size() +} + func (ms *MutableStateImpl) HasRequestID( requestID string, ) bool { @@ -2610,6 +2649,70 @@ func (ms *MutableStateImpl) HasRequestID( return ok } +// closeTransactionSweepChasmRequestIDs bounds the CHASM-attached request IDs (those recorded via +// AttachChasmRequestID) retained per execution for UpdateComponent idempotency. It first evicts every +// entry older than RequestIDMaxAge, then evicts the oldest of the survivors beyond the +// MaximumRequestIDsPerExecution count cap. Sweeping by age first reclaims stale entries in a batch, +// rather than a single oldest entry per transaction once at the cap. +// +// Only entries with a non-nil AttachTime are sweepable; the create request ID and event-backed request +// IDs always have a nil AttachTime and are never swept. Run lazily on transaction close, when a request +// ID has been added. +func (ms *MutableStateImpl) closeTransactionSweepChasmRequestIDs(transactionPolicy historyi.TransactionPolicy) { + if !ms.chasmRequestIDsAdded || transactionPolicy != historyi.TransactionPolicyActive { + return + } + + nsName := ms.namespaceEntry.Name().String() + limit := ms.config.MaximumRequestIDsPerExecution(nsName) + maxAge := ms.config.RequestIDMaxAge(nsName) + if limit <= 0 && maxAge <= 0 { + return // both count cap and age sweep disabled + } + + evicted := 0 + evict := func(id string, info *persistencespb.RequestIDInfo) { + delete(ms.executionState.RequestIds, id) + ms.approximateSize -= chasmRequestIDApproxSize(id, info) + evicted++ + } + + // First sweep entries older than maxAge; the rest are candidates for the count cap. + type sweepable struct { + id string + info *persistencespb.RequestIDInfo + } + candidates := make([]sweepable, 0, len(ms.executionState.RequestIds)) + cutoff := ms.timeSource.Now().Add(-maxAge) + for id, info := range ms.executionState.RequestIds { + if info.GetAttachTime() == nil { + continue // create and event-backed request IDs are never swept + } + if maxAge > 0 && info.GetAttachTime().AsTime().Before(cutoff) { + evict(id, info) + continue + } + candidates = append(candidates, sweepable{id: id, info: info}) + } + + // Enforce hard length limit. + if limit > 0 && len(candidates) > limit { + slices.SortFunc(candidates, func(a, b sweepable) int { + return cmp.Or( + a.info.GetAttachTime().AsTime().Compare(b.info.GetAttachTime().AsTime()), + cmp.Compare(a.id, b.id), + ) + }) + for _, c := range candidates[:len(candidates)-limit] { + evict(c.id, c.info) + } + } + + metrics.CHASMRequestIDEvicted.With( + ms.metricsHandler.WithTags(metrics.NamespaceTag(nsName)), + ).Record(int64(evicted)) +} + func (ms *MutableStateImpl) addWorkflowExecutionStartedEventForContinueAsNew( ctx context.Context, parentExecutionInfo *workflowspb.ParentExecutionInfo, @@ -7739,6 +7842,10 @@ func (ms *MutableStateImpl) closeTransaction( ms.chasmNodeSizes[nodePath] = newSize } + // Enforce the CHASM request-ID limit before the executionState blob is captured into the + // mutation/snapshot below. Gating (active + attached-this-transaction) lives in the helper. + ms.closeTransactionSweepChasmRequestIDs(transactionPolicy) + if isStateDirty { if err := ms.closeTransactionUpdateTransitionHistory( transactionPolicy, @@ -8370,6 +8477,7 @@ func (ms *MutableStateImpl) cleanupTransaction() error { ms.visibilityUpdated = false ms.executionStateUpdated = false ms.workflowTaskUpdated = false + ms.chasmRequestIDsAdded = false ms.isResetStateUpdated = false ms.timeSkippingInfoUpdated = false ms.updateInfoUpdated = make(map[string]struct{}) diff --git a/service/history/workflow/mutable_state_impl_test.go b/service/history/workflow/mutable_state_impl_test.go index b806b54ede..138eb89a5a 100644 --- a/service/history/workflow/mutable_state_impl_test.go +++ b/service/history/workflow/mutable_state_impl_test.go @@ -6458,6 +6458,292 @@ func (s *mutableStateSuite) TestHasRequestID_EmptyExecutionState() { } } +func (s *mutableStateSuite) TestAttachChasmRequestID() { + ts := clock.NewEventTimeSource() + baseTime := time.Now().UTC().Truncate(time.Second) + ts.Update(baseTime) + s.mutableState.timeSource = ts + + // A pre-existing create request ID must never be treated as CHASM-attached. + s.mutableState.AttachRequestID("create-req", enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, 1) + s.Equal("create-req", s.mutableState.executionState.CreateRequestId) + + sizeBefore := s.mutableState.approximateSize + s.mutableState.AttachChasmRequestID("chasm-req") + + info, ok := s.mutableState.executionState.RequestIds["chasm-req"] + s.True(ok) + s.NotNil(info.GetAttachTime(), "CHASM-attached entry must carry an attach time (the sweepable marker)") + s.Equal(baseTime, info.GetAttachTime().AsTime()) + s.Equal(enumspb.EVENT_TYPE_UNSPECIFIED, info.GetEventType()) + s.Zero(info.GetEventId()) + + // CreateRequestId is left untouched. + s.Equal("create-req", s.mutableState.executionState.CreateRequestId) + + // Forces persistence and gates the lazy sweep on this transaction. + s.True(s.mutableState.executionStateUpdated) + s.True(s.mutableState.chasmRequestIDsAdded) + s.Greater(s.mutableState.approximateSize, sizeBefore) + + // An empty request ID is a no-op. + s.mutableState.AttachChasmRequestID("") + _, ok = s.mutableState.executionState.RequestIds[""] + s.False(ok) +} + +func (s *mutableStateSuite) TestCloseTransactionSweepChasmRequestIDs() { + baseTime := time.Now().UTC().Truncate(time.Second) + + // seedPriorRequestIDs records n sweepable CHASM request IDs "prior-00".."prior-(n-1)" directly, + // with strictly increasing attach times, as if attached in earlier (already-closed) transactions. + seedPriorRequestIDs := func(n int) { + if s.mutableState.executionState.RequestIds == nil { + s.mutableState.executionState.RequestIds = make(map[string]*persistencespb.RequestIDInfo) + } + for i := range n { + s.mutableState.executionState.RequestIds[fmt.Sprintf("prior-%02d", i)] = &persistencespb.RequestIDInfo{ + AttachTime: timestamppb.New(baseTime.Add(time.Duration(i) * time.Second)), + } + } + } + // seedPriorAt records a single sweepable CHASM request ID at attach time t, as if attached in an + // earlier (already-closed) transaction. + seedPriorAt := func(id string, t time.Time) { + if s.mutableState.executionState.RequestIds == nil { + s.mutableState.executionState.RequestIds = make(map[string]*persistencespb.RequestIDInfo) + } + s.mutableState.executionState.RequestIds[id] = &persistencespb.RequestIDInfo{ + AttachTime: timestamppb.New(t), + } + } + // attachNow records a request ID via the real path at time t, marking it attached-this-transaction + // (satisfying the sweep gate). + attachNow := func(id string, t time.Time) { + ts := clock.NewEventTimeSource() + ts.Update(t) + s.mutableState.timeSource = ts + s.mutableState.AttachChasmRequestID(id) + } + countSweepable := func() int { + n := 0 + for _, info := range s.mutableState.executionState.RequestIds { + if info.GetAttachTime() != nil { + n++ + } + } + return n + } + // setLimits sets both sweep knobs. Count-focused subtests pass a large maxAge (nothing is old + // enough to be aged out); age-focused subtests pass a small one. Set explicitly in every subtest + // because SetupSubTest resets only the mutable state, not mockConfig, so values leak otherwise. + setLimits := func(count int, maxAge time.Duration) { + s.mockConfig.MaximumRequestIDsPerExecution = func(string) int { return count } + s.mockConfig.RequestIDMaxAge = func(string) time.Duration { return maxAge } + } + const nonTriggeringAge = 7 * 24 * time.Hour + + s.Run("active + attached: count cap evicts oldest, never create/event-backed", func() { + s.SetupSubTest() + setLimits(3, nonTriggeringAge) + s.mutableState.AttachRequestID("create-req", enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, 1) + s.mutableState.AttachRequestID("event-req", enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED, 5) + seedPriorRequestIDs(4) + attachNow("current-req", baseTime.Add(time.Hour)) // newest + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + // limit=3: the 3 newest sweepable entries are kept (current-req plus the two newest priors). + s.False(s.mutableState.HasRequestID("prior-00")) + s.False(s.mutableState.HasRequestID("prior-01")) + s.True(s.mutableState.HasRequestID("prior-02")) + s.True(s.mutableState.HasRequestID("prior-03")) + s.True(s.mutableState.HasRequestID("current-req")) + // Create and event-backed request IDs (nil attach time) are never swept. + s.True(s.mutableState.HasRequestID("create-req")) + s.True(s.mutableState.HasRequestID("event-req")) + s.Equal(3, countSweepable()) + }) + + s.Run("active + attached but at count cap: no sweep", func() { + s.SetupSubTest() + setLimits(3, nonTriggeringAge) + seedPriorRequestIDs(2) // 2 priors + attachNow("current-req", baseTime.Add(time.Hour)) // + 1 attached = 3 sweepable == limit + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + // The gate fires, but the total is at the cap and nothing is aged, so nothing is evicted. + s.True(s.mutableState.HasRequestID("prior-00")) + s.True(s.mutableState.HasRequestID("prior-01")) + s.True(s.mutableState.HasRequestID("current-req")) + s.Equal(3, countSweepable()) + }) + + s.Run("age batch evicts entries below the count cap", func() { + s.SetupSubTest() + setLimits(100, 30*time.Minute) // count cap never binds + seedPriorRequestIDs(4) // attached near baseTime, so older than the cutoff + attachNow("current-req", baseTime.Add(time.Hour)) + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + // cutoff = now(baseTime+1h) - 30m: the priors are older and swept even though we're far under + // the count cap; current-req is younger and kept. + for i := range 4 { + s.False(s.mutableState.HasRequestID(fmt.Sprintf("prior-%02d", i))) + } + s.True(s.mutableState.HasRequestID("current-req")) + s.Equal(1, countSweepable()) + }) + + s.Run("age batch exceeds count overflow: reclaims whole batch, drops below cap", func() { + s.SetupSubTest() + setLimits(3, 30*time.Minute) + seedPriorRequestIDs(6) // all older than the cutoff + attachNow("current-req", baseTime.Add(time.Hour)) + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + // overLimit=7-3=4, aged=6; max(4,6)=6 evicts the whole aged batch, leaving 1 (below the cap). + // This is the point of the age knob: headroom, not one-at-a-time eviction while pinned at the cap. + for i := range 6 { + s.False(s.mutableState.HasRequestID(fmt.Sprintf("prior-%02d", i))) + } + s.True(s.mutableState.HasRequestID("current-req")) + s.Equal(1, countSweepable()) + }) + + s.Run("count overflow exceeds age batch: count cap dominates", func() { + s.SetupSubTest() + setLimits(3, 30*time.Minute) + seedPriorAt("aged-0", baseTime) // older than cutoff + seedPriorAt("aged-1", baseTime.Add(time.Second)) // older than cutoff + seedPriorAt("recent-0", baseTime.Add(40*time.Minute)) // younger than cutoff + seedPriorAt("recent-1", baseTime.Add(41*time.Minute)) // younger than cutoff + seedPriorAt("recent-2", baseTime.Add(42*time.Minute)) // younger than cutoff + attachNow("current-req", baseTime.Add(time.Hour)) + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + // overLimit=6-3=3, aged=2; max(3,2)=3 evicts the 3 oldest - both aged entries plus the oldest + // still-fresh one - to hold the hard cap. + s.False(s.mutableState.HasRequestID("aged-0")) + s.False(s.mutableState.HasRequestID("aged-1")) + s.False(s.mutableState.HasRequestID("recent-0")) + s.True(s.mutableState.HasRequestID("recent-1")) + s.True(s.mutableState.HasRequestID("recent-2")) + s.True(s.mutableState.HasRequestID("current-req")) + s.Equal(3, countSweepable()) + }) + + s.Run("count cap disabled (limit<=0): age still sweeps", func() { + s.SetupSubTest() + setLimits(0, 30*time.Minute) + seedPriorRequestIDs(4) // older than the cutoff + attachNow("current-req", baseTime.Add(time.Hour)) + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + for i := range 4 { + s.False(s.mutableState.HasRequestID(fmt.Sprintf("prior-%02d", i))) + } + s.True(s.mutableState.HasRequestID("current-req")) + s.Equal(1, countSweepable()) + }) + + s.Run("age disabled (maxAge<=0): count cap only", func() { + s.SetupSubTest() + setLimits(100, 0) // count cap never binds, age disabled + seedPriorRequestIDs(4) // old, but age sweeping is off + attachNow("current-req", baseTime.Add(time.Hour)) + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + for i := range 4 { + s.True(s.mutableState.HasRequestID(fmt.Sprintf("prior-%02d", i))) + } + s.True(s.mutableState.HasRequestID("current-req")) + s.Equal(5, countSweepable()) + }) + + s.Run("count cap and age both disabled: no sweep", func() { + s.SetupSubTest() + setLimits(0, 0) + seedPriorRequestIDs(4) + attachNow("current-req", baseTime.Add(time.Hour)) + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + for i := range 4 { + s.True(s.mutableState.HasRequestID(fmt.Sprintf("prior-%02d", i))) + } + s.True(s.mutableState.HasRequestID("current-req")) + }) + + s.Run("passive: no sweep", func() { + s.SetupSubTest() + setLimits(1, 30*time.Minute) + seedPriorRequestIDs(4) + attachNow("current-req", baseTime.Add(time.Hour)) + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyPassive) + + for i := range 4 { + s.True(s.mutableState.HasRequestID(fmt.Sprintf("prior-%02d", i))) + } + s.True(s.mutableState.HasRequestID("current-req")) + }) + + s.Run("active but nothing attached this transaction: no sweep", func() { + s.SetupSubTest() + setLimits(1, 30*time.Minute) + seedPriorRequestIDs(4) // over both limits, but none attached this transaction + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + for i := range 4 { + s.True(s.mutableState.HasRequestID(fmt.Sprintf("prior-%02d", i))) + } + }) + + s.Run("deterministic tie-break on equal attach time", func() { + s.SetupSubTest() + setLimits(3, nonTriggeringAge) + // All priors share one attach time; ties break lexicographically (largest survive). + s.mutableState.executionState.RequestIds = make(map[string]*persistencespb.RequestIDInfo) + for _, id := range []string{"c", "a", "e", "b", "d"} { + s.mutableState.executionState.RequestIds[id] = &persistencespb.RequestIDInfo{ + AttachTime: timestamppb.New(baseTime), + } + } + attachNow("current-req", baseTime.Add(time.Hour)) + + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + // limit=3: current-req is strictly newest; among the equal-time priors the lexicographically + // largest ("d", "e") survive and the rest are evicted. + s.True(s.mutableState.HasRequestID("d")) + s.True(s.mutableState.HasRequestID("e")) + s.False(s.mutableState.HasRequestID("a")) + s.False(s.mutableState.HasRequestID("b")) + s.False(s.mutableState.HasRequestID("c")) + s.True(s.mutableState.HasRequestID("current-req")) + }) + + s.Run("eviction shrinks approximateSize", func() { + s.SetupSubTest() + setLimits(1, nonTriggeringAge) + seedPriorRequestIDs(4) + attachNow("current-req", baseTime.Add(time.Hour)) + + sizeBefore := s.mutableState.approximateSize + s.mutableState.closeTransactionSweepChasmRequestIDs(historyi.TransactionPolicyActive) + + s.Less(s.mutableState.approximateSize, sizeBefore) + }) +} + func (s *mutableStateSuite) TestAddTasks_CHASMPureTask() { s.mockConfig.ChasmMaxInMemoryPureTasks = dynamicconfig.GetIntPropertyFn(5) totalTasks := 2 * s.mockConfig.ChasmMaxInMemoryPureTasks()