Vihren Atlas

workflow_task_state_machine.go ×21

Code frontier

unlabeled · c_7acea99b0811

233 tests · 4692 LOC · 174 files · introduces 0 tests · 136 LOC · 4 files

Introduces — evidence that enters the hierarchy at this concept

Code
26 ranges136 lines · 4 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
803 ranges4692 lines · 174 files · Browse complete extent
All tests (intent)
233 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: 136 introduced LOC across 26 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/workflow/workflow_task_state_machine.go 110 introduced LOC · 21 ranges

Open complete file

179 redirectCounter int64,
180 suggestContinueAsNewReasons []enumspb.SuggestContinueAsNewReason,
181 > ) (*historyi.WorkflowTaskInfo, error) { workflow_task_state_machine.go
182 > // When this function is called from ApplyEvents, workflowTask is nil.
183 > // It is safe to look up the workflow task as it does not have to deal with transient workflow task case.
184 > if workflowTask == nil {
185 workflowTask = m.GetWorkflowTaskByID(scheduledEventID)
186 if workflowTask == nil {
202 }
203
204 > workflowTask = &historyi.WorkflowTaskInfo{ workflow_task_state_machine.go
205 > Version: version,
206 > ScheduledEventID: scheduledEventID,
207 > StartedEventID: startedEventID,
208 > RequestID: requestID,
209 > WorkflowTaskTimeout: workflowTask.WorkflowTaskTimeout,
210 > Attempt: workflowTask.Attempt,
211 > AttemptsSinceLastSuccess: workflowTask.AttemptsSinceLastSuccess,
212 > StartedTime: startedTime,
213 > ScheduledTime: workflowTask.ScheduledTime,
214 > TaskQueue: workflowTask.TaskQueue,
215 > OriginalScheduledTime: workflowTask.OriginalScheduledTime,
216 > Type: workflowTask.Type,
217 > SuggestContinueAsNew: suggestContinueAsNew,
218 > SuggestContinueAsNewReasons: suggestContinueAsNewReasons,
219 > HistorySizeBytes: historySizeBytes,
220 > BuildIdRedirectCounter: redirectCounter,
221 > Stamp: m.ms.GetExecutionInfo().GetWorkflowTaskStamp(),
222 > }
223 >
224 > if buildId := worker_versioning.BuildIdIfUsingVersioning(versioningStamp); buildId != "" {
225 if redirectCounter == 0 {
226 // this is the initial build ID, it should normally be persisted after scheduling the wf task,
462 targetDeploymentVersion *deploymentpb.WorkerDeploymentVersion,
463 targetRevisionNumber int64,
464 > ) (*historypb.HistoryEvent, *historyi.WorkflowTaskInfo, error) { workflow_task_state_machine.go
465 > opTag := tag.WorkflowActionWorkflowTaskStarted
466 > workflowTask := m.GetWorkflowTaskByID(scheduledEventID)
467 > if workflowTask == nil || workflowTask.StartedEventID != common.EmptyEventID {
468 m.ms.logger.Warn(mutableStateInvalidHistoryActionMsg, opTag,
469 tag.WorkflowEventID(m.ms.GetNextEventID()),
473 }
474
475 > m.ms.RemoveSpeculativeWorkflowTaskTimeoutTask() workflow_task_state_machine.go
476 >
477 > scheduledEventID = workflowTask.ScheduledEventID
478 > startedEventID := scheduledEventID + 1
479 > startTime := m.ms.timeSource.Now()
480 >
481 > // The history size computed here might not include this workflow task scheduled or started
482 > // events. That's okay, it doesn't have to be 100% accurate. It just has to be kept
483 > // consistent between the started event in history and the event that was sent to the SDK
484 > // that resulted in the successful completion.
485 > historySizeBytes, suggestContinueAsNewReasons := m.getHistorySizeInfo()
486 > suggestContinueAsNew := len(suggestContinueAsNewReasons) > 0
487 > if updateReg != nil && updateReg.SuggestContinueAsNew() {
488 suggestContinueAsNew = cmp.Or(suggestContinueAsNew, true)
489 suggestContinueAsNewReasons = append(suggestContinueAsNewReasons, enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_UPDATES)
490 }
491
492 > if m.ms.config.EnableSendTargetVersionChanged(m.ms.namespaceEntry.Name().String()) && workflow_task_state_machine.go
493 > m.ms.GetEffectiveVersioningBehavior() != enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {
494
495 // effectiveDeploymentVersion may be nil if the workflow is on an unversioned build;
542 }
543 // emit metric
544 > if m.targetWorkerDeploymentVersionChangedForStartedEvent() { workflow_task_state_machine.go
545 metrics.WorkflowTargetVersionChangedCount.With(m.metricsHandler.WithTags(
546 metrics.NamespaceTag(m.ms.namespaceEntry.Name().String()),
548 )).Record(1)
549 }
550 > if suggestContinueAsNew { workflow_task_state_machine.go
551 metrics.WorkflowSuggestContinueAsNewCount.With(m.metricsHandler.WithTags(
552 metrics.NamespaceTag(m.ms.namespaceEntry.Name().String()),
561 }
562
563 > workflowTask, scheduledEventCreatedForRedirect, redirectCounter, err := m.processBuildIdRedirectInfo(versioningStamp, workflowTask, taskQueue, redirectInfo, skipVersioningCheck) workflow_task_state_machine.go
564 > if err != nil {
565 return nil, nil, err
566 }
567
568 > workflowTaskScheduledEventCreated := scheduledEventCreatedForRedirect || workflow_task_state_machine.go
569 > (!m.ms.IsTransientWorkflowTask() && workflowTask.Type != enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE)
570 >
571 > // If new events came since transient/speculative WT was scheduled or failover happened during lifetime of transient/speculative WT,
572 > // transient/speculative WT needs to be converted to normal WT, i.e. WorkflowTaskScheduledEvent needs to be created now.
573 > if !workflowTaskScheduledEventCreated &&
574 > (workflowTask.ScheduledEventID != m.ms.GetNextEventID() || workflowTask.Version != m.ms.GetCurrentVersion()) {
575
576 workflowTask.Attempt = 1
590 // Create WorkflowTaskStartedEvent only if WorkflowTaskScheduledEvent was created.
591 // (it wasn't created for transient/speculative WT).
592 > var startedEvent *historypb.HistoryEvent workflow_task_state_machine.go
593 > if workflowTaskScheduledEventCreated {
594 startedEvent = m.ms.hBuilder.AddWorkflowTaskStartedEvent(
595 scheduledEventID,
608 }
609
610 > workflowTask, err = m.ApplyWorkflowTaskStartedEvent( workflow_task_state_machine.go
611 > workflowTask,
612 > m.ms.GetCurrentVersion(),
613 > scheduledEventID,
614 > startedEventID,
615 > requestID,
616 > startTime,
617 > suggestContinueAsNew,
618 > historySizeBytes,
619 > versioningStamp,
620 > redirectCounter,
621 > suggestContinueAsNewReasons,
622 > )
623 > if err != nil {
624 return nil, nil, err
625 }
626
627 > m.emitWorkflowTaskAttemptStats(workflowTask.Attempt) workflow_task_state_machine.go
628 >
629 > // TODO merge active & passive task generation
630 > if err = m.ms.taskGenerator.GenerateStartWorkflowTaskTasks(
631 > scheduledEventID,
632 > ); err != nil {
633 return nil, nil, err
634 }
635
636 > return startedEvent, workflowTask, nil workflow_task_state_machine.go
637 }
638
648 redirectInfo *taskqueuespb.BuildIdRedirectInfo,
649 skipVersioningCheck bool,
650 > ) (newWorkflowTask *historyi.WorkflowTaskInfo, converted bool, redirectCounter int64, err error) { workflow_task_state_machine.go
651 > buildId := worker_versioning.BuildIdIfUsingVersioning(versioningStamp)
652 > if buildId == "" && (m.ms.GetAssignedBuildId() == "" || // unversioned workflow
653 > skipVersioningCheck || // resetter may add WFT started events without stamps, it sets skipVersioningCheck=true
654 > (taskQueue.GetKind() == enumspb.TASK_QUEUE_KIND_STICKY && m.ms.executionInfo.GetStickyTaskQueue() == taskQueue.GetName())) {
655 > // build ID is expected to be empty for sticky queues until old versioning is removed [cleanup-old-wv]
656 > return workflowTask, false, 0, nil
657 > }
658
659 redirectCounter, err = m.ms.validateBuildIdRedirectInfo(versioningStamp, redirectInfo)
1155 m.ms.executionInfo.WorkflowTaskAttemptsSinceLastSuccess = workflowTask.AttemptsSinceLastSuccess
1156 if !workflowTask.StartedTime.IsZero() {
1157 > m.ms.executionInfo.WorkflowTaskStartedTime = timestamppb.New(workflowTask.StartedTime) workflow_task_state_machine.go
1158 > }
1159 if !workflowTask.ScheduledTime.IsZero() {
1160 m.ms.executionInfo.WorkflowTaskScheduledTime = timestamppb.New(workflowTask.ScheduledTime)
1449 func (m *workflowTaskStateMachine) emitWorkflowTaskAttemptStats(
1450 attempt int32,
1451 > ) { workflow_task_state_machine.go
1452 > namespaceName := m.ms.GetNamespaceEntry().Name().String()
1453 > metrics.WorkflowTaskAttempt.With(m.ms.metricsHandler).
1454 > Record(int64(attempt), metrics.NamespaceTag(namespaceName))
1455 > if attempt >= int32(m.ms.shard.GetConfig().WorkflowTaskCriticalAttempts()) {
1456 m.ms.shard.GetThrottledLogger().Warn("Critical attempts processing workflow task",
1457 tag.WorkflowNamespace(namespaceName),
1489 }
1490
1491 > func (m *workflowTaskStateMachine) getHistorySizeInfo() (int64, []enumspb.SuggestContinueAsNewReason) { workflow_task_state_machine.go
1492 > var reasons []enumspb.SuggestContinueAsNewReason
1493 > stats := m.ms.GetExecutionInfo().ExecutionStats
1494 > if stats == nil {
1495 return 0, reasons
1496 }
1498 // include the workflow task started event that we're currently writing. That's okay, it
1499 // doesn't have to be 100% accurate.
1500 > historySize := stats.HistorySize workflow_task_state_machine.go
1501 > // This is called right before AddWorkflowTaskStartedEvent, so at this point, nextEventID
1502 > // is the ID of the workflow task started event.
1503 > historyCount := m.ms.GetNextEventID()
1504 > config := m.ms.shard.GetConfig()
1505 > namespaceName := m.ms.GetNamespaceEntry().Name().String()
1506 > sizeLimit := int64(config.HistorySizeSuggestContinueAsNew(namespaceName))
1507 > countLimit := int64(config.HistoryCountSuggestContinueAsNew(namespaceName))
1508 > if historySize >= sizeLimit {
1509 reasons = append(reasons, enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_HISTORY_SIZE_TOO_LARGE)
1510 }
1511 > if historyCount >= countLimit { workflow_task_state_machine.go
1512 reasons = append(reasons, enumspb.SUGGEST_CONTINUE_AS_NEW_REASON_TOO_MANY_HISTORY_EVENTS)
1513 }
1514 > return historySize, reasons workflow_task_state_machine.go
1515 }
1516
go.temporal.io/server/service/history/workflow/task_generator.go 19 introduced LOC · 2 ranges

Open complete file

517 func (r *TaskGeneratorImpl) GenerateStartWorkflowTaskTasks(
518 workflowTaskScheduledEventID int64,
519 > ) error { task_generator.go
520 > workflowTask := r.mutableState.GetWorkflowTaskByID(
521 > workflowTaskScheduledEventID,
522 > )
523 > if workflowTask == nil {
524 return serviceerror.NewInternalf("it could be a bug, cannot get pending workflow task: %v", workflowTaskScheduledEventID)
525 }
526
527 > isSpeculative := workflowTask.Type == enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE task_generator.go
528 > wttt := &tasks.WorkflowTaskTimeoutTask{
529 > // TaskID is set by shard
530 > WorkflowKey: r.mutableState.GetWorkflowKey(),
531 > VisibilityTimestamp: workflowTask.StartedTime.Add(workflowTask.WorkflowTaskTimeout),
532 > TimeoutType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
533 > EventID: workflowTask.ScheduledEventID,
534 > ScheduleAttempt: workflowTask.Attempt,
535 > Version: workflowTask.Version,
536 > Stamp: workflowTask.Stamp,
537 > InMemory: isSpeculative,
538 > }
539 >
540 > if isSpeculative {
541 // If WT is speculative, create task in in-memory task queue.
542 return r.mutableState.SetSpeculativeWorkflowTaskTimeoutTask(wttt)
go.temporal.io/server/service/history/workflow/mutable_state_impl.go 4 introduced LOC · 2 ranges

Open complete file

3538 targetDeploymentVersion *deploymentpb.WorkerDeploymentVersion,
3539 targetRevisionNumber int64,
3540 > ) (*historypb.HistoryEvent, *historyi.WorkflowTaskInfo, error) { mutable_state_impl.go
3541 > opTag := tag.WorkflowActionWorkflowTaskStarted
3542 > if err := ms.checkMutability(opTag); err != nil {
3543 return nil, nil, err
3544 }
3545 > return ms.workflowTaskManager.AddWorkflowTaskStartedEvent(scheduledEventID, requestID, taskQueue, identity, versioningStamp, redirectInfo, skipVersioningCheck, updateReg, targetDeploymentVersion, targetRevisionNumber) mutable_state_impl.go
3546 }
3547
go.temporal.io/server/service/history/tasks/workflow_task_timer.go 3 introduced LOC · 1 range

Open complete file

64 }
65
66 > func (d *WorkflowTaskTimeoutTask) SetVisibilityTime(t time.Time) { workflow_task_timer.go
67 > d.VisibilityTimestamp = t
68 > }
69
70 func (d *WorkflowTaskTimeoutTask) GetCategory() Category {

Direct hierarchy

Parents

Children