Vihren Atlas

workflow.go ×9

Code frontier

unlabeled · c_5c7ff96e591f

33 tests · 3255 LOC · 136 files · introduces 0 tests · 87 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
13 ranges87 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
532 ranges3255 lines · 136 files · Browse complete extent
All tests (intent)
33 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.

2 files ranked by introduced lines: 87 introduced LOC across 13 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/workerdeployment/workflow.go 61 introduced LOC · 9 ranges

Open complete file

1617
1618 // syncUnversionedRamp should not be called in async mode
1619 > func (d *WorkflowRunner) syncUnversionedRamp(ctx workflow.Context, versionUpdateArgs *deploymentspb.SyncVersionStateUpdateArgs) error { workflow.go
1620 > activityCtx := workflow.WithActivityOptions(ctx, defaultActivityOptions)
1621 >
1622 > // DescribeVersion activity to get all the task queues in the current version, or the ramping version if current is nil
1623 > version := d.State.RoutingConfig.CurrentVersion //nolint:staticcheck // SA1019: worker versioning v0.31
1624 > if version == worker_versioning.UnversionedVersionId {
1625 version = d.State.RoutingConfig.RampingVersion //nolint:staticcheck // SA1019: worker versioning v0.31
1626 }
1627
1628 > if d.rampingVersionStringUnversioned(version) { workflow.go
1629 return nil
1630 }
1631
1632 > var res deploymentspb.DescribeVersionFromWorkerDeploymentActivityResult workflow.go
1633 > err := workflow.ExecuteActivity(
1634 > activityCtx,
1635 > d.a.DescribeVersionFromWorkerDeployment,
1636 > &deploymentspb.DescribeVersionFromWorkerDeploymentActivityArgs{
1637 > Version: version,
1638 > }).Get(ctx, &res)
1639 > if err != nil {
1640 return err
1641 }
1642
1643 // send in the task-queue families in batches of syncBatchSize
1644 > batches := make([][]*deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData, 0) workflow.go
1645 > syncReqs := make([]*deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData, 0)
1646 >
1647 > // Grouping by task-queue name
1648 > taskQueuesByName := make(map[string][]enumspb.TaskQueueType)
1649 > for _, tq := range res.GetTaskQueueInfos() {
1650 > taskQueuesByName[tq.GetName()] = append(taskQueuesByName[tq.GetName()], tq.GetType())
1651 > }
1652
1653 > for _, tqName := range workflow.DeterministicKeys(taskQueuesByName) { workflow.go
1654 > tqTypes := taskQueuesByName[tqName]
1655 > sync := &deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData{
1656 > Name: tqName,
1657 > Types: tqTypes,
1658 > Data: &deploymentspb.DeploymentVersionData{
1659 > Version: nil,
1660 > RoutingUpdateTime: versionUpdateArgs.RoutingUpdateTime,
1661 > RampingSinceTime: versionUpdateArgs.RampingSinceTime,
1662 > RampPercentage: versionUpdateArgs.RampPercentage,
1663 > },
1664 > }
1665 > syncReqs = append(syncReqs, sync)
1666 >
1667 > if len(syncReqs) == int(d.State.SyncBatchSize) {
1668 batches = append(batches, syncReqs)
1669 syncReqs = make([]*deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData, 0) // reset the syncReq.Sync slice for the next batch
1670 }
1671 }
1672 > if len(syncReqs) > 0 { workflow.go
1673 > batches = append(batches, syncReqs)
1674 > }
1675
1676 // calling SyncDeploymentVersionUserData for each batch
1677 > for _, batch := range batches { workflow.go
1678 > var syncRes deploymentspb.SyncDeploymentVersionUserDataResponse
1679 >
1680 > err = workflow.ExecuteActivity(activityCtx, d.a.SyncDeploymentVersionUserDataFromWorkerDeployment, &deploymentspb.SyncDeploymentVersionUserDataRequest{
1681 > DeploymentName: d.DeploymentName,
1682 > Version: nil,
1683 > ForgetVersion: false,
1684 > Sync: batch,
1685 > }).Get(ctx, &syncRes)
1686 > if err != nil {
1687 // TODO (Shivam): Compensation functions required to roll back the local state + activity changes.
1688 return err
1689 }
1690
1691 > if len(syncRes.TaskQueueMaxVersions) > 0 { workflow.go
1692 > // wait for propagation
1693 > err = workflow.ExecuteActivity(
1694 > activityCtx,
1695 > d.a.CheckUnversionedRampUserDataPropagation,
1696 > &deploymentspb.CheckWorkerDeploymentUserDataPropagationRequest{
1697 > TaskQueueMaxVersions: syncRes.TaskQueueMaxVersions,
1698 > }).Get(ctx, nil)
1699 > if err != nil {
1700 // TODO (Shivam): Compensation functions required to roll back the local state + activity changes.
1701 return err
1704 }
1705
1706 > return err workflow.go
1707 }
1708
go.temporal.io/server/api/deployment/v1/message.pb.go 26 introduced LOC · 4 ranges

Open complete file

1170 func (*DescribeVersionFromWorkerDeploymentActivityArgs) ProtoMessage() {}
1171
1172 > func (x *DescribeVersionFromWorkerDeploymentActivityArgs) ProtoReflect() protoreflect.Message { message.pb.go
1173 > mi := &file_temporal_server_api_deployment_v1_message_proto_msgTypes[12]
1174 > if x != nil {
1175 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1176 > if ms.LoadMessageInfo() == nil {
1177 > ms.StoreMessageInfo(mi)
1178 > }
1179 > return ms
1180 }
1181 return mi.MessageOf(x)
1202 }
1203
1204 > func (x *DescribeVersionFromWorkerDeploymentActivityResult) Reset() { message.pb.go
1205 > *x = DescribeVersionFromWorkerDeploymentActivityResult{}
1206 > mi := &file_temporal_server_api_deployment_v1_message_proto_msgTypes[13]
1207 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1208 > ms.StoreMessageInfo(mi)
1209 > }
1210
1211 func (x *DescribeVersionFromWorkerDeploymentActivityResult) String() string {
1215 func (*DescribeVersionFromWorkerDeploymentActivityResult) ProtoMessage() {}
1216
1217 > func (x *DescribeVersionFromWorkerDeploymentActivityResult) ProtoReflect() protoreflect.Message { message.pb.go
1218 > mi := &file_temporal_server_api_deployment_v1_message_proto_msgTypes[13]
1219 > if x != nil {
1220 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1221 > if ms.LoadMessageInfo() == nil {
1222 > ms.StoreMessageInfo(mi)
1223 > }
1224 > return ms
1225 }
1226 return mi.MessageOf(x)
1232 }
1233
1234 > func (x *DescribeVersionFromWorkerDeploymentActivityResult) GetTaskQueueInfos() []*v11.WorkerDeploymentVersionInfo_VersionTaskQueueInfo { message.pb.go
1235 > if x != nil {
1236 > return x.TaskQueueInfos
1237 > }
1238 return nil
1239 }

Direct hierarchy

Parents

Children