Vihren Atlas

transfer_queue_active_task_executor.go ×18

Code frontier

unlabeled · c_37f40bad493b

18 tests · 7504 LOC · 229 files · introduces 0 tests · 53 LOC · 1 file

Introduces — evidence that enters the hierarchy at this concept

Code
18 ranges53 lines · 1 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
1509 ranges7504 lines · 229 files · Browse complete extent
All tests (intent)
18 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.

1 file ranked by introduced lines: 53 introduced LOC across 18 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/transfer_queue_active_task_executor.go 53 introduced LOC · 18 ranges

Open complete file

159 case *tasks.WorkflowTask:
160 err = t.processWorkflowTask(ctx, task)
161 > case *tasks.CloseExecutionTask: transfer_queue_active_task_executor.go
162 > err = t.processCloseExecution(ctx, task)
163 case *tasks.CancelExecutionTask:
164 err = t.processCancelExecution(ctx, task)
375 ctx context.Context,
376 task *tasks.CloseExecutionTask,
377 > ) (retError error) { transfer_queue_active_task_executor.go
378 > ctx, cancel := context.WithTimeout(ctx, taskTimeout)
379 > defer cancel()
380 >
381 > weContext, release, err := getWorkflowExecutionContextForTask(ctx, t.shardContext, t.cache, task)
382 > if err != nil {
383 return err
384 }
385 > defer func() { release(retError) }() transfer_queue_active_task_executor.go
386
387 > mutableState, err := loadMutableStateForTransferTask(ctx, t.shardContext, weContext, task, t.metricHandler, t.logger) transfer_queue_active_task_executor.go
388 > if err != nil {
389 return err
390 }
391 > if mutableState == nil || mutableState.IsWorkflowExecutionRunning() { transfer_queue_active_task_executor.go
392 return nil
393 }
395 // DeleteAfterClose is set to true when this close execution task was generated as part of delete open workflow execution procedure.
396 // Delete workflow execution is started by user API call and should be done regardless of current workflow version.
397 > if !task.DeleteAfterClose { transfer_queue_active_task_executor.go
398 > closeVersion, err := mutableState.GetCloseVersion()
399 > if err != nil {
400 return err
401 }
402 > err = CheckTaskVersion(t.shardContext, t.logger, mutableState.GetNamespaceEntry(), closeVersion, task.Version, task) transfer_queue_active_task_executor.go
403 > if err != nil {
404 return err
405 }
406 }
407
408 > workflowExecution := commonpb.WorkflowExecution{ transfer_queue_active_task_executor.go
409 > WorkflowId: task.GetWorkflowID(),
410 > RunId: task.GetRunID(),
411 > }
412 > executionInfo := mutableState.GetExecutionInfo()
413 > children := copyChildWorkflowInfos(mutableState.GetPendingChildExecutionInfos())
414 > var completionEvent *historypb.HistoryEvent // needed to report close event to parent workflow
415 > replyToParentWorkflow := mutableState.HasParentExecution() && executionInfo.NewExecutionRunId == ""
416 > if replyToParentWorkflow || len(children) > 0 {
417 // only load close event if needed.
418 completionEvent, err = mutableState.GetCompletionEvent(ctx)
422 replyToParentWorkflow = replyToParentWorkflow && !ndc.IsTerminatedByResetter(completionEvent)
423 }
424 > parentNamespaceID := executionInfo.ParentNamespaceId transfer_queue_active_task_executor.go
425 > parentWorkflowID := executionInfo.ParentWorkflowId
426 > parentRunID := executionInfo.ParentRunId
427 > parentInitiatedID := executionInfo.ParentInitiatedId
428 > parentInitiatedVersion := executionInfo.ParentInitiatedVersion
429 > var parentClock *clockspb.VectorClock
430 > if executionInfo.ParentClock != nil {
431 parentClock = vclock.NewVectorClock(
432 executionInfo.ParentClock.ClusterId,
436 }
437
438 > namespaceName := mutableState.GetNamespaceEntry().Name() transfer_queue_active_task_executor.go
439 >
440 > firstRunID, err := mutableState.GetFirstRunID(ctx)
441 > if err != nil {
442 return err
443 }
446 // Release lock immediately since mutable state is not needed
447 // and the rest of logic is RPC calls, which can take time.
448 > release(nil) transfer_queue_active_task_executor.go
449 >
450 > // Communicate the result to parent execution if this is Child Workflow execution
451 > if replyToParentWorkflow {
452 _, err := t.historyRawClient.RecordChildExecutionCompleted(ctx, &historyservice.RecordChildExecutionCompletedRequest{
453 NamespaceId: parentNamespaceID,
478 // So we need to additionally check the termination reason for this parent to determine if this task was indeed created due to reset or due to normal completion of the WF.
479 // Also, checking the dynamic config is not strictly safe since by definition it can change at any time. However this reduces the chance of us skipping the parent close policy when we shouldn't.
480 > allowResetWithPendingChildren := t.config.AllowResetWithPendingChildren(namespaceName.String()) transfer_queue_active_task_executor.go
481 > shouldSkipParentClosePolicy := false
482 > isParentTerminatedDueToReset := (completionEvent != nil) && ndc.IsTerminatedByResetter(completionEvent)
483 > if isParentTerminatedDueToReset && executionInfo.GetResetRunId() != "" && allowResetWithPendingChildren {
484 // TODO (Chetan): update this condition as new reset policies/cases are added.
485 shouldSkipParentClosePolicy = true // only skip if the parent is reset and we are using the new flow.
486 }
487 > if !shouldSkipParentClosePolicy { transfer_queue_active_task_executor.go
488 if err := t.processParentClosePolicy(
489 ctx,

Direct hierarchy

Parents

Children