workflow_task_state_machine.go ×7

Frontier kind: Code frontier

unlabeled · c_89aaa90be990

295 tests · 4203 LOC · 169 files · introduces 0 tests · 93 LOC · 4 files

Introduces — evidence that enters the hierarchy at this concept

Code
14 ranges93 lines · 4 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
683 ranges4203 lines · 169 files · Browse complete extent
All tests (intent)
295 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

4 files ranked by introduced lines: 93 introduced LOC across 14 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/workflow/workflow_task_state_machine.go 73 introduced LOC · 7 ranges

Open complete file

69 originalScheduledTimestamp *timestamppb.Timestamp,
70 workflowTaskType enumsspb.WorkflowTaskType,
71 > ) (*historyi.WorkflowTaskInfo, error) { workflow_task_state_machine.go
72 >
73 > // set workflow state to running, since workflow task is scheduled
74 > // NOTE: for zombie workflow, should not change the state
75 > state, _ := m.ms.GetWorkflowStateStatus()
76 > if state != enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE {
77 > if _, err := m.ms.UpdateWorkflowStateStatus(
78 > enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING,
79 > enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
80 > ); err != nil {
81 return nil, err
82 }
83 }
84
85 > workflowTask := &historyi.WorkflowTaskInfo{ workflow_task_state_machine.go
86 > Version: version,
87 > ScheduledEventID: scheduledEventID,
88 > StartedEventID: common.EmptyEventID,
89 > RequestID: emptyUUID,
90 > WorkflowTaskTimeout: startToCloseTimeout.AsDuration(),
91 > TaskQueue: taskQueue,
92 > Attempt: attempt,
93 > AttemptsSinceLastSuccess: m.ms.executionInfo.WorkflowTaskAttemptsSinceLastSuccess,
94 > ScheduledTime: scheduledTime.AsTime(),
95 > StartedTime: time.Time{},
96 > OriginalScheduledTime: originalScheduledTimestamp.AsTime(),
97 > Type: workflowTaskType,
98 > SuggestContinueAsNew: false, // reset, will be recomputed on workflow task started
99 > SuggestContinueAsNewReasons: nil, // reset, will be recomputed on workflow task started
100 > HistorySizeBytes: 0, // reset, will be recomputed on workflow task started
101 > Stamp: m.ms.GetExecutionInfo().GetWorkflowTaskStamp(),
102 > }
103 >
104 > m.retainWorkflowTaskBuildIdInfo(workflowTask)
105 > m.UpdateWorkflowTask(workflowTask)
106 > return workflowTask, nil
107 }
108
111 // - BuildIdRedirectCounter so add the right BuildIdRedirectCounter to the WFT started event that will be
112 // created at WFT completion time
113 > func (m *workflowTaskStateMachine) retainWorkflowTaskBuildIdInfo(workflowTask *historyi.WorkflowTaskInfo) { workflow_task_state_machine.go
114 > if workflowTask.Attempt > 1 {
115 workflowTask.BuildId = m.ms.executionInfo.WorkflowTaskBuildId
116 workflowTask.BuildIdRedirectCounter = m.ms.executionInfo.BuildIdRedirectCounter
1129 func (m *workflowTaskStateMachine) UpdateWorkflowTask(
1130 workflowTask *historyi.WorkflowTaskInfo,
1132 > if m.HasStartedWorkflowTask() && workflowTask.StartedEventID == common.EmptyEventID {
1133 // reset the flag whenever started workflow task closes, there could be three cases:
1134 // 1. workflow task completed:
1147 }
1148
1149 > m.ms.executionInfo.WorkflowTaskVersion = workflowTask.Version workflow_task_state_machine.go
1150 > m.ms.executionInfo.WorkflowTaskScheduledEventId = workflowTask.ScheduledEventID
1151 > m.ms.executionInfo.WorkflowTaskStartedEventId = workflowTask.StartedEventID
1152 > m.ms.executionInfo.WorkflowTaskRequestId = workflowTask.RequestID
1153 > m.ms.executionInfo.WorkflowTaskTimeout = durationpb.New(workflowTask.WorkflowTaskTimeout)
1154 > m.ms.executionInfo.WorkflowTaskAttempt = workflowTask.Attempt
1155 > m.ms.executionInfo.WorkflowTaskAttemptsSinceLastSuccess = workflowTask.AttemptsSinceLastSuccess
1156 > if !workflowTask.StartedTime.IsZero() {
1157 m.ms.executionInfo.WorkflowTaskStartedTime = timestamppb.New(workflowTask.StartedTime)
1158 }
1159 > if !workflowTask.ScheduledTime.IsZero() { workflow_task_state_machine.go
1160 > m.ms.executionInfo.WorkflowTaskScheduledTime = timestamppb.New(workflowTask.ScheduledTime)
1161 > }
1162 > m.ms.executionInfo.WorkflowTaskOriginalScheduledTime = timestamppb.New(workflowTask.OriginalScheduledTime)
1163 > m.ms.executionInfo.WorkflowTaskType = workflowTask.Type
1164 > m.ms.executionInfo.WorkflowTaskSuggestContinueAsNew = workflowTask.SuggestContinueAsNew
1165 > m.ms.executionInfo.WorkflowTaskSuggestContinueAsNewReasons = workflowTask.SuggestContinueAsNewReasons
1166 > m.ms.executionInfo.WorkflowTaskHistorySizeBytes = workflowTask.HistorySizeBytes
1167 > m.ms.executionInfo.WorkflowTaskBuildId = workflowTask.BuildId
1168 > m.ms.executionInfo.WorkflowTaskBuildIdRedirectCounter = workflowTask.BuildIdRedirectCounter
1169 >
1170 > m.ms.workflowTaskUpdated = true
1171 >
1172 > // NOTE:
1173 > // - do not update executionInfo.TaskQueue!
1174 >
1175 > m.ms.logger.Debug("Workflow task updated",
1176 > tag.WorkflowScheduledEventID(workflowTask.ScheduledEventID),
1177 > tag.WorkflowStartedEventID(workflowTask.StartedEventID),
1178 > tag.WorkflowTaskRequestId(workflowTask.RequestID),
1179 > tag.WorkflowTaskTimeout(workflowTask.WorkflowTaskTimeout),
1180 > tag.Attempt(workflowTask.Attempt),
1181 > tag.WorkflowStartedTimestamp(workflowTask.StartedTime),
1182 > tag.WorkflowTaskType(workflowTask.Type.String()))
1183 }
1184
1215
1216 // GetWorkflowTaskByID returns details about the current workflow task by scheduled event ID.
1217 > func (m *workflowTaskStateMachine) GetWorkflowTaskByID(scheduledEventID int64) *historyi.WorkflowTaskInfo { workflow_task_state_machine.go
1218 > workflowTask := m.getWorkflowTaskInfo()
1219 > if scheduledEventID == workflowTask.ScheduledEventID {
1220 > return workflowTask
1221 > }
1222
1223 return nil
go.temporal.io/server/common/log/tag/tags.go 15 introduced LOC · 5 ranges

Open complete file

169
170 // WorkflowTaskTimeout returns tag for WorkflowTaskTimeoutSeconds
171 > func WorkflowTaskTimeout(s time.Duration) ZapTag { tags.go
172 > return NewDurationTag("workflow-task-timeout", s)
173 > }
174
175 // QueryID returns tag for QueryID
214
215 // WorkflowStartedEventID returns tag for WorkflowStartedEventID
216 > func WorkflowStartedEventID(startedEventID int64) ZapTag { tags.go
217 > return NewInt64("wf-started-event-id", startedEventID)
218 > }
219
220 // WorkflowStartedTimestamp returns tag for WorkflowStartedTimestamp
221 > func WorkflowStartedTimestamp(t time.Time) ZapTag { tags.go
222 > return NewTimeTag("wf-started-timestamp", t)
223 > }
224
225 // WorkflowInitiatedID returns tag for WorkflowInitiatedID
683 }
684
685 > func WorkflowTaskType(wtType string) ZapTag { tags.go
686 > return NewStringTag("wt-type", wtType)
687 > }
688
689 // AttemptCount returns tag for AttemptCount
934
935 // WorkflowTaskRequestId returns a tag for workflow task RequestId
936 > func WorkflowTaskRequestId(s string) ZapTag { tags.go
937 > return NewStringTag("workflow-task-request-id", s)
938 > }
939
940 // AckLevel returns tag for ack level
go.temporal.io/server/service/history/workflow/mutable_state_impl.go 3 introduced LOC · 1 range

Open complete file

2336
2337 // GetWorkflowTaskByID returns details about the current workflow task by scheduled event ID.
2338 > func (ms *MutableStateImpl) GetWorkflowTaskByID(scheduledEventID int64) *historyi.WorkflowTaskInfo { mutable_state_impl.go
2339 > return ms.workflowTaskManager.GetWorkflowTaskByID(scheduledEventID)
2340 > }
2341
2342 func (ms *MutableStateImpl) GetPendingActivityInfos() map[int64]*persistencespb.ActivityInfo {
go.temporal.io/server/api/enums/v1/workflow_task_type.pb.go 2 introduced LOC · 1 range

Open complete file

56 }
57
58 > func (x WorkflowTaskType) String() string { workflow_task_type.pb.go
59 > switch x {
60 case WORKFLOW_TASK_TYPE_UNSPECIFIED:
61 return "Unspecified"