Vihren Atlas

workflow.go ×16

Code frontier

unlabeled · c_be106c64ded1

62 tests · 2989 LOC · 133 files · introduces 0 tests · 71 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
17 ranges71 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
466 ranges2989 lines · 133 files · Browse complete extent
All tests (intent)
62 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: 71 introduced LOC across 17 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/workerdeployment/workflow.go 65 introduced LOC · 16 ranges

Open complete file

746 }
747
748 > func (d *WorkflowRunner) validateStateBeforeAcceptingRampingUpdate(args *deploymentspb.SetRampingVersionArgs) error { workflow.go
749 > //nolint:staticcheck // SA1019: worker versioning v0.31
750 > if args.Version == d.State.GetRoutingConfig().GetRampingVersion() &&
751 > args.Percentage == d.State.GetRoutingConfig().GetRampingVersionPercentage() &&
752 > args.Identity == d.State.GetLastModifierIdentity() {
753 return temporal.NewApplicationError("version already ramping, no change", errNoChangeType, d.State.GetConflictToken())
754 }
755
756 > if args.ConflictToken != nil && !bytes.Equal(args.ConflictToken, d.State.GetConflictToken()) { workflow.go
757 return temporal.NewApplicationError("conflict token mismatch", errFailedPrecondition)
758 }
759
760 //nolint:staticcheck // SA1019: worker versioning v0.31
761 > if args.Version == d.State.GetRoutingConfig().GetCurrentVersion() && workflow.go
762 > !(args.Version == worker_versioning.UnversionedVersionId && args.Percentage == 0) {
763 d.logger.Info("version can't be set to ramping since it is already current")
764 return temporal.NewApplicationError(fmt.Sprintf("requested ramping version %s is already current", args.Version), errFailedPrecondition)
765 }
766
767 > if _, ok := d.State.GetVersions()[args.Version]; !ok && workflow.go
768 > args.Version != worker_versioning.UnversionedVersionId &&
769 > args.Version != "" &&
770 > !args.GetAllowNoPollers() {
771 d.logger.Info("version not found in deployment")
772 return temporal.NewApplicationError(fmt.Sprintf("requested ramping version %s not found in deployment", args.Version), errVersionNotFound)
773 }
774
775 > if d.State.ManagerIdentity != "" && d.State.ManagerIdentity != args.Identity { workflow.go
776 return serviceerror.NewFailedPrecondition(fmt.Sprintf(ErrManagerIdentityMismatch, d.State.ManagerIdentity, args.Identity))
777 }
778 > return nil workflow.go
779 }
780
788 //revive:disable-next-line:cognitive-complexity
789 //nolint:staticcheck // deprecated stuff will be cleaned
790 > func (d *WorkflowRunner) handleSetRampingVersion(ctx workflow.Context, args *deploymentspb.SetRampingVersionArgs) (*deploymentspb.SetRampingVersionResponse, error) { workflow.go
791 > if err := d.preUpdateChecks(ctx); err != nil {
792 return nil, err
793 }
794
795 // use lock to enforce only one update at a time
796 > err := d.lock.Lock(ctx) workflow.go
797 > if err != nil {
798 d.logger.Error("Could not acquire workflow lock")
799 return nil, serviceerror.NewDeadlineExceeded("Could not acquire workflow lock")
800 }
801 > defer func() { workflow.go
802 > // Even if the update doesn't change the state we mark it as dirty because of created history events.
803 > d.setStateChanged()
804 > d.lock.Unlock()
805 > }()
806
807 // Validating the state before starting the SetRampingVersion operation. This is required due to the following reason:
810 // enforces sequential updates, after the first update completes, the local state of the deployment workflow will change. The second update,
811 // now already accepted by the validator, should now not be allowed to run since the state of the workflow is different.
812 > err = d.validateStateBeforeAcceptingRampingUpdate(args) workflow.go
813 > if err != nil {
814 return nil, err
815 }
816
817 > prevRampingVersion := d.State.RoutingConfig.RampingVersion workflow.go
818 > prevRampingVersionPercentage := d.State.RoutingConfig.RampingVersionPercentage
819 >
820 > newRampingVersion := args.Version
821 > routingUpdateTime := timestamppb.New(workflow.Now(ctx))
822 >
823 > if _, ok := d.State.Versions[args.Version]; !ok &&
824 > args.Version != worker_versioning.UnversionedVersionId &&
825 > args.Version != "" &&
826 > args.GetAllowNoPollers() {
827 d.logger.Info("version not found in deployment, but AllowNoPollers is true, so we will create the version")
828 if err := d.addVersionToWorkerDeployment(ctx, &deploymentspb.AddVersionUpdateArgs{Version: newRampingVersion, CreateTime: routingUpdateTime}); err != nil {
842 }
843
844 > var rampingSinceTime *timestamppb.Timestamp workflow.go
845 > var rampingVersionUpdateTime *timestamppb.Timestamp
846 >
847 > asyncMode := d.hasMinVersion(AsyncSetCurrentAndRamping)
848 >
849 > if prevRampingVersion != newRampingVersion || prevRampingVersionPercentage != args.Percentage || !asyncMode {
850 > // In async mode we do not touch routing config and versions if this is not changing the
851 > // ramping version or percentage (but could still come here because modifier identity is changing).
852 >
853 > // Determine timestamps based on whether we're setting or unsetting ramp
854 > if newRampingVersion == "" {
855 // unsetting ramp
856 rampingVersionUpdateTime = routingUpdateTime
857 > } else if prevRampingVersion == newRampingVersion { workflow.go
858 // version was already ramping, user changing ramp %
859 rampingSinceTime = d.State.RoutingConfig.RampingVersionChangedTime
868 // Initialize for both sync and async modes to simplify state update logic
869 // Ensure CurrentDeploymentVersion is populated from deprecated field if needed for backward compatibility
870 > currentDeploymentVersion := d.State.RoutingConfig.CurrentDeploymentVersion workflow.go
871 > if currentDeploymentVersion == nil {
872 currentDeploymentVersion = worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(d.State.RoutingConfig.CurrentVersion)
873 }
874 > pendingRoutingConfig := &deploymentpb.RoutingConfig{ workflow.go
875 > CurrentDeploymentVersion: currentDeploymentVersion,
876 > CurrentVersion: d.State.RoutingConfig.CurrentVersion,
877 > RampingDeploymentVersion: worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(newRampingVersion),
878 > RampingVersion: newRampingVersion,
879 > RampingVersionPercentage: args.Percentage,
880 > CurrentVersionChangedTime: d.State.RoutingConfig.CurrentVersionChangedTime,
881 > RampingVersionChangedTime: rampingVersionUpdateTime,
882 > RampingVersionPercentageChangedTime: routingUpdateTime,
883 > RevisionNumber: d.State.RoutingConfig.RevisionNumber,
884 > }
885 >
886 > var routingConfigToSync *deploymentpb.RoutingConfig
887 >
888 > if asyncMode {
889 pendingRoutingConfig.RevisionNumber++
890 // only setting it in the request if it's async mode
892 }
893
894 > if newRampingVersion == "" { workflow.go
895 err = d.unsetRamp(ctx, routingUpdateTime, routingConfigToSync, prevRampingVersion, asyncMode, pendingRoutingConfig)
896 if err != nil {
go.temporal.io/server/api/deployment/v1/message.pb.go 6 introduced LOC · 1 range

Open complete file

2938 }
2939
2940 > func (x *SetRampingVersionArgs) Reset() { message.pb.go
2941 > *x = SetRampingVersionArgs{}
2942 > mi := &file_temporal_server_api_deployment_v1_message_proto_msgTypes[43]
2943 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
2944 > ms.StoreMessageInfo(mi)
2945 > }
2946
2947 func (x *SetRampingVersionArgs) String() string {

Direct hierarchy

Parents

Children