Vihren Atlas

mutable_state_impl.go ×44

Code frontier

unlabeled · c_9d57a1abdb80

196 tests · 5219 LOC · 177 files · introduces 0 tests · 241 LOC · 4 files

Introduces — evidence that enters the hierarchy at this concept

Code
68 ranges241 lines · 4 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
940 ranges5219 lines · 177 files · Browse complete extent
All tests (intent)
196 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: 241 introduced LOC across 68 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/workflow/mutable_state_impl.go 127 introduced LOC · 44 ranges

Open complete file

3607 eventID int64,
3608 maxResetPoints int,
3609 > ) bool { mutable_state_impl.go
3610 > resetPoints := ms.executionInfo.AutoResetPoints.GetPoints()
3611 > for _, rp := range resetPoints {
3612 if rp.GetBinaryChecksum() == binaryChecksum && rp.GetBuildId() == buildId {
3613 return false
3617 // todo@time-skipping: time skipping is naturally supported for auto reset points, and need to
3618 // decide if this the best default policy for auto reset points
3619 > newPoint := &workflowpb.ResetPointInfo{ mutable_state_impl.go
3620 > BinaryChecksum: binaryChecksum,
3621 > BuildId: buildId,
3622 > RunId: ms.executionState.GetRunId(),
3623 > FirstWorkflowTaskCompletedId: eventID,
3624 > CreateTime: timestamppb.New(ms.timeSource.Now()),
3625 > Resettable: ms.CheckResettable() == nil,
3626 > }
3627 > ms.executionInfo.AutoResetPoints = &workflowpb.ResetPoints{
3628 > Points: util.SliceTail(append(resetPoints, newPoint), maxResetPoints),
3629 > }
3630 > return true
3631 }
3632
3762 usedVersion *deploymentpb.WorkerDeploymentVersion,
3763 maxSearchAttributeValueSize int,
3764 > ) error { mutable_state_impl.go
3765 > changed, err := ms.addBuildIDAndDeploymentInfoToSearchAttributesWithNoVisibilityTask(stamp, usedVersion, maxSearchAttributeValueSize)
3766 > if err != nil {
3767 return err
3768 }
3769
3770 > if !changed { mutable_state_impl.go
3771 return nil
3772 }
3773 > return ms.taskGenerator.GenerateUpsertVisibilityTask() mutable_state_impl.go
3774 }
3775
3776 > func (ms *MutableStateImpl) loadBuildIds() ([]string, error) { mutable_state_impl.go
3777 > searchAttributes := ms.executionInfo.SearchAttributes
3778 > if searchAttributes == nil {
3779 > return []string{}, nil
3780 > }
3781 saPayload, found := searchAttributes[sadefs.BuildIds]
3782 if !found {
3797 }
3798
3799 > func (ms *MutableStateImpl) loadSearchAttributeString(saName string) (string, error) { mutable_state_impl.go
3800 > searchAttributes := ms.executionInfo.SearchAttributes
3801 > if searchAttributes == nil {
3802 > return "", nil
3803 > }
3804 saPayload, found := searchAttributes[saName]
3805 if !found {
3820 }
3821
3822 > func (ms *MutableStateImpl) loadUsedDeploymentVersions() ([]string, error) { mutable_state_impl.go
3823 > searchAttributes := ms.executionInfo.SearchAttributes
3824 > if searchAttributes == nil {
3825 > return []string{}, nil
3826 > }
3827 saPayload, found := searchAttributes[sadefs.TemporalUsedWorkerDeploymentVersions]
3828 if !found {
3851 existingValues []string,
3852 stamp *commonpb.WorkerVersionStamp,
3853 > ) []string { mutable_state_impl.go
3854 > var newValues []string
3855 > var buildId string
3856 >
3857 > behavior := ms.GetWorkflowVersioningBehaviorSA()
3858 >
3859 > // set up the unversioned or assigned:x sentinels (versioning v2)
3860 > if !stamp.GetUseVersioning() && behavior == enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED { // unversioned workflows may still have non-nil deployment, so we don't check deployment
3861 > newValues = append(newValues, worker_versioning.UnversionedSearchAttribute)
3862 > } else if ms.GetAssignedBuildId() != "" && behavior == enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {
3863 newValues = append(newValues, worker_versioning.AssignedBuildIdSearchAttribute(ms.GetAssignedBuildId()))
3864 }
3865
3866 // get the most up-to-date pinned entry put it at the front (v3 reachability and v3.1 drainage)
3867 > if behavior == enumspb.VERSIONING_BEHAVIOR_PINNED { mutable_state_impl.go
3868 newValues = append(newValues, worker_versioning.PinnedBuildIdSearchAttribute(ms.GetWorkerDeploymentVersionSA()))
3869 }
3870
3871 // get the build id entry (all versions of versioning)
3872 > if stamp != nil { mutable_state_impl.go
3873 buildId = worker_versioning.VersionStampToBuildIdSearchAttribute(stamp)
3874 }
3875
3876 // add all previous values except for unversioned, assigned, or pinned (there can only be one, and we just added it)
3877 > foundBuildId := false mutable_state_impl.go
3878 > for _, existingValue := range existingValues {
3879 if existingValue == buildId {
3880 foundBuildId = true
3887
3888 // add buildId to the list only if it wasn't there before
3889 > if !foundBuildId && buildId != "" { mutable_state_impl.go
3890 newValues = append(newValues, buildId)
3891 }
3892 > return newValues mutable_state_impl.go
3893 }
3894
3895 > func (ms *MutableStateImpl) addUsedDeploymentVersionToLoadedSearchAttribute(existingValues []string, usedVersion *deploymentpb.WorkerDeploymentVersion) []string { mutable_state_impl.go
3896 > if usedVersion == nil {
3897 > return existingValues
3898 > }
3899
3900 // Get the current deployment version string (already formatted via ExternalWorkerDeploymentVersionToString)
3918 }
3919
3920 > func (ms *MutableStateImpl) saveBuildIds(buildIds []string, maxSearchAttributeValueSize int) error { mutable_state_impl.go
3921 > searchAttributes := ms.executionInfo.SearchAttributes
3922 > if searchAttributes == nil {
3923 > searchAttributes = make(map[string]*commonpb.Payload, 1)
3924 > ms.executionInfo.SearchAttributes = searchAttributes
3925 > }
3926
3927 > hasUnversionedOrAssigned := false mutable_state_impl.go
3928 > if len(buildIds) > 0 { // len is 0 if we are removing the pinned search attribute and the workflow was never unversioned or assigned
3929 > hasUnversionedOrAssigned = worker_versioning.IsUnversionedOrAssignedBuildIdSearchAttribute(buildIds[0])
3930 > }
3931 > for {
3932 > saPayload, err := sadefs.EncodeValue(buildIds, enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST)
3933 > if err != nil {
3934 return err
3935 }
3936 > if len(buildIds) == 0 || len(saPayload.GetData()) <= maxSearchAttributeValueSize { mutable_state_impl.go
3937 > ms.updateSearchAttributes(map[string]*commonpb.Payload{sadefs.BuildIds: saPayload})
3938 > break
3939 }
3940 if len(buildIds) == 1 {
4025 usedVersion *deploymentpb.WorkerDeploymentVersion,
4026 maxSearchAttributeValueSize int,
4027 > ) (bool, error) { mutable_state_impl.go
4028 > // get all the existing SAs
4029 > existingBuildIds, err := ms.loadBuildIds()
4030 > if err != nil {
4031 return false, err
4032 }
4033 > existingUsedDeploymentVersions, err := ms.loadUsedDeploymentVersions() mutable_state_impl.go
4034 > if err != nil {
4035 return false, err
4036 }
4037 > existingDeployment, err := ms.loadSearchAttributeString(sadefs.TemporalWorkerDeployment) mutable_state_impl.go
4038 > if err != nil {
4039 return false, err
4040 }
4041 > existingVersion, err := ms.loadSearchAttributeString(sadefs.TemporalWorkerDeploymentVersion) mutable_state_impl.go
4042 > if err != nil {
4043 return false, err
4044 }
4045 > existingBehavior, err := ms.loadSearchAttributeString(sadefs.TemporalWorkflowVersioningBehavior) mutable_state_impl.go
4046 > if err != nil {
4047 return false, err
4048 }
4049
4050 // modify them
4051 > modifiedBuildIds := ms.addBuildIdToLoadedSearchAttribute(existingBuildIds, stamp) mutable_state_impl.go
4052 > modifiedUsedDeploymentVersions := ms.addUsedDeploymentVersionToLoadedSearchAttribute(existingUsedDeploymentVersions, usedVersion)
4053 > modifiedDeployment := ms.GetWorkerDeploymentSA()
4054 > modifiedVersion := ms.GetWorkerDeploymentVersionSA()
4055 > modifiedBehavior := ""
4056 > if b := ms.GetWorkflowVersioningBehaviorSA(); b != enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {
4057 modifiedBehavior = b.String()
4058 }
4059
4060 // check equality
4061 > if slices.Equal(existingBuildIds, modifiedBuildIds) && mutable_state_impl.go
4062 > slices.Equal(existingUsedDeploymentVersions, modifiedUsedDeploymentVersions) &&
4063 > existingDeployment == modifiedDeployment &&
4064 > existingVersion == modifiedVersion &&
4065 > existingBehavior == modifiedBehavior {
4066 return false, nil
4067 }
4068
4069 // save build ids if changed
4070 > if !slices.Equal(existingBuildIds, modifiedBuildIds) { mutable_state_impl.go
4071 > err = ms.saveBuildIds(modifiedBuildIds, maxSearchAttributeValueSize)
4072 > if err != nil {
4073 return false, err // if err != nil, nothing will be written
4074 }
4076
4077 // save used deployment versions if changed
4078 > if !slices.Equal(existingUsedDeploymentVersions, modifiedUsedDeploymentVersions) { mutable_state_impl.go
4079 err = ms.saveUsedDeploymentVersions(modifiedUsedDeploymentVersions, maxSearchAttributeValueSize)
4080 if err != nil {
4084
4085 // save deployment search attributes if changed
4086 > if existingDeployment != modifiedDeployment || mutable_state_impl.go
4087 > existingVersion != modifiedVersion ||
4088 > existingBehavior != modifiedBehavior {
4089 err = ms.saveDeploymentSearchAttributes(modifiedDeployment, modifiedVersion, modifiedBehavior, maxSearchAttributeValueSize)
4090 if err != nil {
4098
4099 // CheckResettable check if workflow can be reset
4100 > func (ms *MutableStateImpl) CheckResettable() error { mutable_state_impl.go
4101 > if len(ms.GetPendingChildExecutionInfos()) > 0 {
4102 return serviceerror.NewInvalidArgument("it is not allowed resetting to a point that workflow has pending child workflow.")
4103 }
4104 > if len(ms.GetPendingRequestCancelExternalInfos()) > 0 { mutable_state_impl.go
4105 return serviceerror.NewInvalidArgument("it is not allowed resetting to a point that workflow has pending request cancel.")
4106 }
4107 > if len(ms.GetPendingSignalExternalInfos()) > 0 { mutable_state_impl.go
4108 return serviceerror.NewInvalidArgument("it is not allowed resetting to a point that workflow has pending signals to send.")
4109 }
4110 > return nil mutable_state_impl.go
4111 }
4112
7043 }
7044
7045 > func (ms *MutableStateImpl) RemoveReportedProblemsSearchAttribute() error { mutable_state_impl.go
7046 > if ms.executionInfo.SearchAttributes == nil {
7047 return nil
7048 }
7049
7050 > temporalReportedProblems := ms.executionInfo.SearchAttributes[sadefs.TemporalReportedProblems] mutable_state_impl.go
7051 > if temporalReportedProblems == nil {
7052 > return nil
7053 > }
7054
7055 // Log the removal of the search attribute
9607 }
9608
9609 > func (ms *MutableStateImpl) GetWorkerDeploymentSA() string { mutable_state_impl.go
9610 > versioningInfo := ms.GetExecutionInfo().GetVersioningInfo()
9611 > if override := versioningInfo.GetVersioningOverride(); override != nil {
9612 if v := worker_versioning.GetOverrideTargetDeploymentVersion(override); v != nil {
9613 return v.GetDeploymentName()
9614 }
9615 }
9616 > if v := versioningInfo.GetDeploymentVersion(); v != nil { mutable_state_impl.go
9617 return v.GetDeploymentName()
9618 }
9619 > return ms.GetExecutionInfo().GetWorkerDeploymentName() mutable_state_impl.go
9620 }
9621
9622 > func (ms *MutableStateImpl) GetWorkerDeploymentVersionSA() string { mutable_state_impl.go
9623 > versioningInfo := ms.GetExecutionInfo().GetVersioningInfo()
9624 > if override := versioningInfo.GetVersioningOverride(); override != nil {
9625 if v := worker_versioning.GetOverrideTargetDeploymentVersion(override); v != nil {
9626 return worker_versioning.ExternalWorkerDeploymentVersionToString(v)
9627 }
9628 }
9629 > if v := versioningInfo.GetDeploymentVersion(); v != nil { mutable_state_impl.go
9630 return worker_versioning.ExternalWorkerDeploymentVersionToString(v)
9631 }
9632 //nolint:staticcheck // SA1019: worker versioning v0.31
9633 > return worker_versioning.ExternalWorkerDeploymentVersionToString(worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(versioningInfo.GetVersion())) mutable_state_impl.go
9634 }
9635
9636 > func (ms *MutableStateImpl) GetWorkflowVersioningBehaviorSA() enumspb.VersioningBehavior { mutable_state_impl.go
9637 > if override := ms.executionInfo.GetVersioningInfo().GetVersioningOverride(); override != nil {
9638 if override.GetAutoUpgrade() {
9639 return enumspb.VERSIONING_BEHAVIOR_AUTO_UPGRADE
9644 return override.GetBehavior()
9645 }
9646 > return ms.executionInfo.GetVersioningInfo().GetBehavior() mutable_state_impl.go
9647 }
9648
go.temporal.io/server/service/history/workflow/workflow_task_state_machine.go 100 introduced LOC · 18 ranges

Open complete file

776 }
777
778 > if !workflowTaskScheduledStartedEventsCreated { workflow_task_state_machine.go
779 // Create corresponding WorkflowTaskScheduled and WorkflowTaskStarted events for transient/speculative workflow tasks.
780 scheduledEvent := m.ms.hBuilder.AddWorkflowTaskScheduledEvent(
810 }
811
812 > deploymentName := request.GetDeploymentOptions().GetDeploymentName() workflow_task_state_machine.go
813 > if deploymentName == "" {
814 > //nolint:staticcheck // SA1019 deprecated Deployment will clean up later
815 > deploymentName = request.GetDeployment().GetSeriesName()
816 > }
817
818 > vb := request.VersioningBehavior workflow_task_state_machine.go
819 > if request.DeploymentOptions != nil && request.DeploymentOptions.GetWorkerVersioningMode() != enumspb.WORKER_VERSIONING_MODE_VERSIONED {
820 // SDK has a bug that reports behavior if user has specified a default behavior without enabling versioning.
821 // Until that is fixed, we should adjust this value so the workflow works correctly.
824
825 //nolint:staticcheck // SA1019 deprecated Deployment will clean up later
826 > wftDeployment := worker_versioning.DeploymentOrVersion(request.Deployment, worker_versioning.DeploymentVersionFromOptions(request.DeploymentOptions)) workflow_task_state_machine.go
827 >
828 > // Now write the completed event
829 > event := m.ms.hBuilder.AddWorkflowTaskCompletedEvent(
830 > workflowTask.ScheduledEventID,
831 > workflowTask.StartedEventID,
832 > request.Identity,
833 > request.BinaryChecksum,
834 > request.WorkerVersionStamp,
835 > request.SdkMetadata,
836 > request.MeteringMetadata,
837 > deploymentName,
838 > wftDeployment,
839 > vb,
840 > )
841 >
842 > override := m.ms.GetExecutionInfo().GetVersioningInfo().GetVersioningOverride()
843 > // Capture the pending one-time target before afterAddWorkflowTaskCompletedEvent,
844 > // which clears the override when this WFT completes on the target version.
845 > var oneTimeTarget *deploymentpb.WorkerDeploymentVersion
846 > if oneTime := override.GetOneTime(); oneTime != nil {
847 oneTimeTarget = oneTime.GetTargetDeploymentVersion()
848 }
849
850 > wftScheduleToClose := event.GetEventTime().AsTime().Sub(workflowTask.ScheduledTime) workflow_task_state_machine.go
851 > err := m.afterAddWorkflowTaskCompletedEvent(event, limits, wftScheduleToClose)
852 > if err != nil {
853 return nil, err
854 }
855 // afterAddWorkflowTaskCompletedEvent clears the one-time override. Emit fulfillment
856 // telemetry only on this live completion path, not when applying rebuilt history.
857 > if oneTimeTarget != nil && wftCompletedOnTargetVersion(wftDeployment, oneTimeTarget) { workflow_task_state_machine.go
858 metrics.WorkerDeploymentVersioningOneTimeOverrideCounter.With(m.metricsHandler).Record(1)
859 m.ms.logger.Info("One-time versioning override fulfilled",
865 }
866
867 > metrics.WorkflowTasksCompleted.With(m.metricsHandler).Record(1, workflow_task_state_machine.go
868 > metrics.NamespaceTag(m.ms.GetNamespaceEntry().Name().String()),
869 > metrics.VersioningBehaviorTag(vb),
870 > metrics.FirstAttemptTag(workflowTask.Attempt),
871 > )
872 >
873 > numConsecutiveWorkflowTaskProblemsToTriggerSearchAttribute := m.ms.config.NumConsecutiveWorkflowTaskProblemsToTriggerSearchAttribute(m.ms.GetNamespaceEntry().Name().String())
874 > if numConsecutiveWorkflowTaskProblemsToTriggerSearchAttribute > 0 {
875 > if err := m.ms.RemoveReportedProblemsSearchAttribute(); err != nil {
876 return nil, err
877 }
878 }
879
880 > return event, nil workflow_task_state_machine.go
881 }
882
1320 limits historyi.WorkflowTaskCompletionLimits,
1321 wftScheduleToClose time.Duration,
1322 > ) error { workflow_task_state_machine.go
1323 > attrs := event.GetWorkflowTaskCompletedEventAttributes()
1324 > m.ms.executionInfo.LastCompletedWorkflowTaskStartedEventId = attrs.GetStartedEventId()
1325 > m.ms.executionInfo.MostRecentWorkerVersionStamp = attrs.GetWorkerVersion()
1326 > m.ms.executionInfo.WorkerDeploymentName = attrs.GetWorkerDeploymentName()
1327 >
1328 > //nolint:staticcheck // SA1019 deprecated Deployment will clean up later
1329 > wftDeployment := attrs.GetDeployment()
1330 > if v := attrs.GetWorkerDeploymentVersion(); v != "" { //nolint:staticcheck // SA1019: worker versioning v0.31
1331 dv, _ := worker_versioning.WorkerDeploymentVersionFromStringV31(v)
1332 wftDeployment = worker_versioning.DeploymentFromDeploymentVersion(dv)
1333 }
1334 > if v := attrs.GetDeploymentVersion(); v != nil { workflow_task_state_machine.go
1335 wftDeployment = worker_versioning.DeploymentFromExternalDeploymentVersion(v)
1336 }
1337 > wftBehavior := attrs.GetVersioningBehavior() workflow_task_state_machine.go
1338 > versioningInfo := m.ms.GetExecutionInfo().GetVersioningInfo()
1339 > transition := m.ms.GetDeploymentTransition()
1340 >
1341 > var completedTransition bool
1342 > if transition != nil {
1343 // It's possible that the completed WFT is not yet from the current transition because when
1344 // the transition started, the current wft was already started. In this case, we allow the
1353
1354 // Deployment and behavior before applying the data came from the completed wft.
1355 > wfDeploymentBefore := m.ms.GetEffectiveDeployment() workflow_task_state_machine.go
1356 > wfBehaviorBefore := m.ms.GetEffectiveVersioningBehavior()
1357 >
1358 > // Change deployment and behavior based on completed wft.
1359 > if wftBehavior == enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {
1360 > if versioningInfo != nil {
1361 versioningInfo.Behavior = wftBehavior
1362 // Deployment Version is not set for unversioned workers.
1380 versioningInfo.DeploymentVersion = worker_versioning.ExternalWorkerDeploymentVersionFromDeployment(wftDeployment)
1381 }
1382 > if oneTime := versioningInfo.GetVersioningOverride().GetOneTime(); oneTime != nil && workflow_task_state_machine.go
1383 > wftCompletedOnTargetVersion(wftDeployment, oneTime.GetTargetDeploymentVersion()) {
1384 // Clear before computing effective deployment/behavior so the worker-reported base state takes effect.
1385 versioningInfo.VersioningOverride = nil
1387
1388 // Deployment and behavior after applying the data came from the completed wft.
1389 > wfDeploymentAfter := m.ms.GetEffectiveDeployment() workflow_task_state_machine.go
1390 > wfBehaviorAfter := m.ms.GetEffectiveVersioningBehavior()
1391 > // We reschedule activities if a transition was completed because during the transition
1392 > // ATs might have been dropped. Note that it is possible that transition completes and still
1393 > // `wfDeploymentBefore == wfDeploymentAfter`. Example: wf was on deployment1, started
1394 > // transition to deployment2, before completing the transition it changed the transition to
1395 > // deployment1 (maybe user rolled back current deployment), now the transition completes.
1396 > if completedTransition ||
1397 > // It is possible that this WFT is changing workflow's deployment even if there was no
1398 > // ongoing transition in the MS. That is possible when the wft is speculative. We still
1399 > // want to reschedule the activities so they are queued with the up-to-date directive.
1400 > !wfDeploymentBefore.Equal(wfDeploymentAfter) ||
1401 > // If effective behavior changes we also want to reschedule the pending activities, so
1402 > // they go to the right matching queues.
1403 > wfBehaviorBefore != wfBehaviorAfter {
1404 if err := m.ms.reschedulePendingActivities(wftScheduleToClose); err != nil {
1405 return err
1408
1409 //nolint:staticcheck // SA1019: worker versioning v2
1410 > buildId := attrs.GetWorkerVersion().GetBuildId() workflow_task_state_machine.go
1411 > if wftDeployment != nil {
1412 buildId = wftDeployment.GetBuildId()
1413 }
1414 > addedResetPoint := m.ms.addResetPointFromCompletion( workflow_task_state_machine.go
1415 > attrs.GetBinaryChecksum(),
1416 > buildId,
1417 > event.GetEventId(),
1418 > limits.MaxResetPoints,
1419 > )
1420 >
1421 > // For v3 versioned workflows (ms.GetEffectiveVersioningBehavior() != UNSPECIFIED), this will update the reachability
1422 > // search attribute based on the execution_info.deployment and/or override deployment if one exists. We must update the
1423 > // search attribute here because the reachability deployment may have just been changed by CompleteDeploymentTransition.
1424 > // This is also useful for unversioned workers.
1425 > // For v1 and v2 versioned workflows the search attributes should be already up-to-date based on the task started events.
1426 > //nolint:staticcheck // SA1019
1427 > if err := m.ms.updateBuildIdsAndDeploymentSearchAttributes(attrs.GetWorkerVersion(), worker_versioning.ExternalWorkerDeploymentVersionFromDeployment(wftDeployment), limits.MaxSearchAttributeValueSize); err != nil {
1428 return err
1429 }
1430 > if addedResetPoint && len(attrs.GetBinaryChecksum()) > 0 { workflow_task_state_machine.go
1431 if err := m.ms.updateBinaryChecksumSearchAttribute(); err != nil {
1432 return err
1433 }
1434 }
1435 > return nil workflow_task_state_machine.go
1436 }
1437
go.temporal.io/server/common/worker_versioning/worker_versioning.go 10 introduced LOC · 5 ranges

Open complete file

114
115 // IsUnversionedOrAssignedBuildIdSearchAttribute returns the value is "unversioned" or "assigned:<bld>"
116 > func IsUnversionedOrAssignedBuildIdSearchAttribute(buildId string) bool { worker_versioning.go
117 > return buildId == UnversionedSearchAttribute ||
118 > strings.HasPrefix(buildId, buildIdSearchAttributePrefixAssigned+BuildIdSearchAttributeDelimiter)
119 > }
120
121 // VersionedBuildIdSearchAttribute returns the search attribute value for a versioned build ID
244 // DeploymentOrVersion Temporary helper function to return a Deployment based on passed Deployment
245 // or WorkerDeploymentVersion objects, if `v` is not nil, it'll take precedence.
246 > func DeploymentOrVersion(d *deploymentpb.Deployment, v *deploymentspb.WorkerDeploymentVersion) *deploymentpb.Deployment { worker_versioning.go
247 > if v != nil {
248 return DeploymentIfValid(DeploymentFromDeploymentVersion(v))
249 }
250 > return DeploymentIfValid(d) worker_versioning.go
251 }
252
253 // DeploymentIfValid returns the deployment back if is both of its fields have value.
254 > func DeploymentIfValid(d *deploymentpb.Deployment) *deploymentpb.Deployment { worker_versioning.go
255 > if d.GetSeriesName() != "" && d.GetBuildId() != "" {
256 return d
257 }
258 > return nil worker_versioning.go
259 }
260
go.temporal.io/server/api/persistence/v1/executions.pb.go 4 introduced LOC · 1 range

Open complete file

1075 }
1076
1077 > func (x *WorkflowExecutionInfo) GetWorkerDeploymentName() string { executions.pb.go
1078 > if x != nil {
1079 > return x.WorkerDeploymentName
1080 > }
1081 return ""
1082 }

Direct hierarchy

Parents

Children