Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

[CHASM] Support WithRequestID on UpdateComponent (#11169)

Browse Source 
## 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 <fred.tzeng@temporal.io>
This commit is contained in:
Lina Jodoin
2026-08-13 14:51:11 -07:00
committed by GitHub
parent 78ee704f37
commit 06e5531513
14 changed files with 966 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

238
api/persistence/v1/executions.pb.go
View File

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

63
chasm/chasmtest/test_engine.go
View File

@@ -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

87
chasm/chasmtest/test_engine_test.go
View File

@@ -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

16
chasm/engine.go
View File

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

12
common/dynamicconfig/constants.go
View File

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

4
common/metrics/metric_defs.go
View File

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

10
proto/internal/temporal/server/api/persistence/v1/executions.proto
View File

@@ -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<string, RequestIDInfo> 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

40
service/history/chasm_engine.go
View File

@@ -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

224
service/history/chasm_engine_test.go
View File

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

4
service/history/configs/config.go
View File

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

3
service/history/interfaces/mutable_state.go
View File

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

12
service/history/interfaces/mutable_state_mock.go
View File

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

110
service/history/workflow/mutable_state_impl.go
View File

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

286
service/history/workflow/mutable_state_impl_test.go
View File

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