Vihren Atlas

transfer_queue_task_executor_base.go ×4

Code frontier

unlabeled · c_0cac5d4b36bf

15 tests · 6243 LOC · 212 files · introduces 0 tests · 53 LOC · 4 files

Introduces — evidence that enters the hierarchy at this concept

Code
7 ranges53 lines · 4 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
1182 ranges6243 lines · 212 files · Browse complete extent
All tests (intent)
15 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: 53 introduced LOC across 7 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/transfer_queue_task_executor_base.go 20 introduced LOC · 4 ranges

Open complete file

153 priority *commonpb.Priority,
154 transactionPolicy historyi.TransactionPolicy,
155 > ) error { transfer_queue_task_executor_base.go
156 > var sst *durationpb.Duration
157 > if workflowTaskScheduleToStartTimeout > 0 {
158 sst = durationpb.New(workflowTaskScheduleToStartTimeout)
159 }
160 > resp, err := t.matchingRawClient.AddWorkflowTask(ctx, &matchingservice.AddWorkflowTaskRequest{ transfer_queue_task_executor_base.go
161 > NamespaceId: task.NamespaceID,
162 > Execution: &commonpb.WorkflowExecution{
163 > WorkflowId: task.WorkflowID,
164 > RunId: task.RunID,
165 > },
166 > TaskQueue: taskqueue,
167 > ScheduledEventId: task.ScheduledEventID,
168 > ScheduleToStartTimeout: sst,
169 > Clock: vclock.NewVectorClock(t.shardContext.GetClusterMetadata().GetClusterID(), t.shardContext.GetShardID(), task.TaskID),
170 > VersionDirective: directive,
171 > Priority: priority,
172 > Stamp: task.Stamp,
173 > })
174 > if _, isNotFound := err.(*serviceerror.NotFound); isNotFound {
175 // NotFound error is not expected for AddTasks calls
176 // but will be ignored by task error handling logic, so log it here
go.temporal.io/server/service/history/workflow/mutable_state_impl.go 17 introduced LOC · 1 range

Open complete file

9747 }
9748
9749 > func (ms *MutableStateImpl) GetShouldUseRampingVersion() bool { mutable_state_impl.go
9750 > execInfo := ms.GetExecutionInfo()
9751 > versioningInfo := execInfo.GetVersioningInfo()
9752 >
9753 > hasNoCompletedWorkflowTask := execInfo.GetLastCompletedWorkflowTaskStartedEventId() == common.EmptyEventID
9754 >
9755 > // Backward-compat note: history events written before ContinueAsNewInitialVersioningBehavior existed
9756 > // carry UNSPECIFIED, which does not match USE_RAMPING_VERSION, so those runs correctly fall through
9757 > // and return false — the same as AUTO_UPGRADE behavior.
9758 > hasInitialBehaviorUseRampingVersion := versioningInfo.GetContinueAsNewInitialVersioningBehavior() == enumspb.CONTINUE_AS_NEW_VERSIONING_BEHAVIOR_USE_RAMPING_VERSION
9759 >
9760 > // ContinueAsNewInitialVersioningBehavior is only populated (in ApplyWorkflowExecutionStartedEvent)
9761 > // from InheritedAutoUpgradeInfo, which always sets Behavior to AUTO_UPGRADE. So there is no world
9762 > // in which ContinueAsNewInitialVersioningBehavior is set without Behavior also being AUTO_UPGRADE.
9763 > // We could also check if the behavior is AUTO_UPGRADE, but that's redundant.
9764 > return hasNoCompletedWorkflowTask && hasInitialBehaviorUseRampingVersion
9765 > }
9766
9767 // reschedulePendingActivities reschedules all the activities that are not started, so they are
go.temporal.io/server/service/history/worker_versioning_util.go 12 introduced LOC · 1 range

Open complete file

198 }
199
200 > func MakeDirectiveForWorkflowTask(ms historyi.MutableState) *taskqueuespb.TaskVersionDirective { worker_versioning_util.go
201 > return worker_versioning.MakeDirectiveForWorkflowTask(
202 > ms.GetInheritedBuildId(),
203 > ms.GetAssignedBuildId(),
204 > ms.GetMostRecentWorkerVersionStamp(),
205 > ms.HasCompletedAnyWorkflowTask(),
206 > ms.GetEffectiveVersioningBehavior(),
207 > ms.GetEffectiveDeployment(),
208 > ms.GetVersioningRevisionNumber(),
209 > ms.GetShouldUseRampingVersion(),
210 > )
211 > }
212
213 func MakeDirectiveForActivityTask(mutableState historyi.MutableState, activityInfo *persistencespb.ActivityInfo) *taskqueuespb.TaskVersionDirective {
go.temporal.io/server/api/persistence/v1/executions.pb.go 4 introduced LOC · 1 range

Open complete file

508 }
509
510 > func (x *WorkflowExecutionInfo) GetLastCompletedWorkflowTaskStartedEventId() int64 { executions.pb.go
511 > if x != nil {
512 > return x.LastCompletedWorkflowTaskStartedEventId
513 > }
514 return 0
515 }

Direct hierarchy

Parents

Children