worker_versioning

package
v1.32.1-x.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 16, 2026 License: MIT Imports: 28 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	BuildIdSearchAttributePrefixPinned = "pinned"

	BuildIdSearchAttributeDelimiter = ":"
	BuildIdSearchAttributeEscape    = "|"
	// UnversionedSearchAttribute is the sentinel value used to mark all unversioned workflows
	UnversionedSearchAttribute = buildIdSearchAttributePrefixUnversioned
	UnversionedVersionId       = "__unversioned__"

	// ErrPinnedVersionNotInTaskQueueSubstring is the key substring used to identify
	// when a pinned version is not present in a task queue. This is used for error
	// classification in batch operations.
	ErrPinnedVersionNotInTaskQueueSubstring = "is not present in task queue"

	// WorkerDeploymentVersionIdDelimiterV31 will be deleted once we stop supporting v31 version string fields
	// in external and internal APIs. Until then, both delimiters are banned in deployment name. All
	// deprecated version string fields in APIs keep using the old delimiter. Workflow SA uses new delimiter.
	WorkerDeploymentVersionIDDelimiterV31   = "."
	WorkerDeploymentVersionDelimiter        = ":"
	WorkerDeploymentVersionWorkflowIDEscape = "|"

	// Prefixes, Delimeters and Keys that are used in the internal entity workflows backing worker-versioning
	WorkerDeploymentWorkflowIDPrefix             = "temporal-sys-worker-deployment"
	WorkerDeploymentVersionWorkflowIDPrefix      = "temporal-sys-worker-deployment-version"
	WorkerDeploymentVersionWorkflowIDInitialSize = len(WorkerDeploymentVersionWorkflowIDPrefix) + len(WorkerDeploymentVersionDelimiter) // 39
	WorkerDeploymentNameFieldName                = "WorkerDeploymentName"
	WorkerDeploymentBuildIDFieldName             = "BuildID"

	// SignalSyncValidationStatus is sent by the WCI workflow to the version workflow
	// when ValidationStatus changes, so the deployment workflow can maintain an
	// up-to-date connectivity summary in its memo.
	SignalSyncValidationStatus = "sync-validation-status"
)

Variables

This section is empty.

Functions

func AddV31VersioningInfoToV32 

func AddV31VersioningInfoToV32(info *workflowpb.WorkflowExecutionVersioningInfo) *workflowpb.WorkflowExecutionVersioningInfo

We store versioning info in the modern v0.32 format, so call this before returning the object to readers to mutatively populate the missing fields.

func AssignedBuildIdSearchAttribute 

func AssignedBuildIdSearchAttribute(buildId string) string

AssignedBuildIdSearchAttribute returns the search attribute value for the currently assigned build ID

func BuildIDToStringV32 

func BuildIDToStringV32(deploymentName, buildID string) string

func BuildIdFromCapabilities 

func BuildIdFromCapabilities(capabilities *commonpb.WorkerVersionCapabilities, options *deploymentpb.WorkerDeploymentOptions) string

func BuildIdIfUsingVersioning 

func BuildIdIfUsingVersioning(stamp *commonpb.WorkerVersionStamp) string

BuildIdIfUsingVersioning returns the given WorkerVersionStamp if it is using versioning, otherwise returns nil.

func CalculateTaskQueueVersioningInfo 

func CalculateTaskQueueVersioningInfo(deployments *persistencespb.DeploymentData) (
	*deploymentspb.WorkerDeploymentVersion,
	int64,
	time.Time,
	*deploymentspb.WorkerDeploymentVersion,
	bool,
	float32,
	int64,
	time.Time,
)

CalculateTaskQueueVersioningInfo calculates the current and ramping versioning info for a task queue.

func CleanupOldDeletedVersions 

func CleanupOldDeletedVersions(deploymentData *persistencespb.WorkerDeploymentData, maxVersions int) bool

CleanupOldDeletedVersions removes versions deleted more than 7 days ago. Also removes more deleted versions if the limit is being exceeded. Never removes undeleted versions. Deprecated. Versions now are deleted serially without using the deleted flag in versionData. TODO: remove this cleanup logic after next major release.

func ConvertOverrideToV32 

func ConvertOverrideToV32(override *workflowpb.VersioningOverride) *workflowpb.VersioningOverride

ConvertOverrideToV32 reads from deprecated fields and returns a new object with ONLY the equivalent non-deprecated v0.32 fields. Should be used to replace any passed in override that is stored in persistence.

func CountDeploymentVersions 

func CountDeploymentVersions(deployments *persistencespb.DeploymentData) int

func DeploymentFromCapabilities 

func DeploymentFromCapabilities(capabilities *commonpb.WorkerVersionCapabilities, options *deploymentpb.WorkerDeploymentOptions) (*deploymentpb.Deployment, error)

DeploymentFromCapabilities returns the deployment if it is using versioning V3, otherwise nil. It returns the deployment from the `options` if present, otherwise, from `capabilities`,

func DeploymentFromDeploymentVersion 

func DeploymentFromDeploymentVersion(dv *deploymentspb.WorkerDeploymentVersion) *deploymentpb.Deployment

DeploymentFromDeploymentVersion Temporary helper function to convert WorkerDeploymentVersion to Deployment proto until we update code to use the new proto in all places.

func DeploymentFromExternalDeploymentVersion 

func DeploymentFromExternalDeploymentVersion(dv *deploymentpb.WorkerDeploymentVersion) *deploymentpb.Deployment

DeploymentFromExternalDeploymentVersion Temporary helper function to convert WorkerDeploymentVersion to Deployment proto until we update code to use the new proto in all places.

func DeploymentIfValid 

func DeploymentIfValid(d *deploymentpb.Deployment) *deploymentpb.Deployment

DeploymentIfValid returns the deployment back if is both of its fields have value.

func DeploymentNameFromCapabilities 

func DeploymentNameFromCapabilities(capabilities *commonpb.WorkerVersionCapabilities, options *deploymentpb.WorkerDeploymentOptions) string

func DeploymentOrVersion 

func DeploymentOrVersion(d *deploymentpb.Deployment, v *deploymentspb.WorkerDeploymentVersion) *deploymentpb.Deployment

DeploymentOrVersion Temporary helper function to return a Deployment based on passed Deployment or WorkerDeploymentVersion objects, if `v` is not nil, it'll take precedence.

func DeploymentVersionFromDeployment 

func DeploymentVersionFromDeployment(deployment *deploymentpb.Deployment) *deploymentspb.WorkerDeploymentVersion

DeploymentVersionFromDeployment Temporary helper function to convert Deployment to WorkerDeploymentVersion proto until we update code to use the new proto in all places.

func DeploymentVersionFromOptions 

func DeploymentVersionFromOptions(options *deploymentpb.WorkerDeploymentOptions) *deploymentspb.WorkerDeploymentVersion

func DirectiveDeployment 

func DirectiveDeployment(directive *taskqueuespb.TaskVersionDirective) *deploymentpb.Deployment

DirectiveDeployment Temporary function until Directive proto is removed.

func ExternalWorkerDeploymentVersionFromDeployment 

func ExternalWorkerDeploymentVersionFromDeployment(deployment *deploymentpb.Deployment) *deploymentpb.WorkerDeploymentVersion

ExternalWorkerDeploymentVersionFromDeployment Temporary helper function to convert Deployment to WorkerDeploymentVersion proto until we update code to use the new proto in all places.

func ExternalWorkerDeploymentVersionFromStringV31 

func ExternalWorkerDeploymentVersionFromStringV31(s string) *deploymentpb.WorkerDeploymentVersion

func ExternalWorkerDeploymentVersionFromVersion 

func ExternalWorkerDeploymentVersionFromVersion(version *deploymentspb.WorkerDeploymentVersion) *deploymentpb.WorkerDeploymentVersion

ExternalWorkerDeploymentVersionFromVersion Temporary helper function to convert internal Worker Deployment to WorkerDeploymentVersion proto until we update code to use the new proto in all places.

func ExternalWorkerDeploymentVersionToString 

func ExternalWorkerDeploymentVersionToString(v *deploymentpb.WorkerDeploymentVersion) string

func ExternalWorkerDeploymentVersionToStringV31 

func ExternalWorkerDeploymentVersionToStringV31(v *deploymentpb.WorkerDeploymentVersion) string

func ExtractVersioningBehaviorFromOverride 

func ExtractVersioningBehaviorFromOverride(override *workflowpb.VersioningOverride) enumspb.VersioningBehavior

func FindBuildId 

func FindBuildId(versioningData *persistencespb.VersioningData, buildId string) (setIndex, indexInSet int)

FindBuildId finds a build ID in the version data's sets, returning (set index, index within that set). Returns -1, -1 if not found.

func FindOldDeploymentVersion 

func FindOldDeploymentVersion(deployments *persistencespb.DeploymentData, v *deploymentspb.WorkerDeploymentVersion) int

func FindTargetDeploymentVersionAndRevisionNumberForWorkflowID 

func FindTargetDeploymentVersionAndRevisionNumberForWorkflowID(
	current *deploymentspb.WorkerDeploymentVersion,
	currentRevisionNumber int64,
	ramping *deploymentspb.WorkerDeploymentVersion,
	rampingPercentage float32,
	rampingRevisionNumber int64,
	workflowId string,
	useRampingVersion bool,
) (*deploymentspb.WorkerDeploymentVersion, int64)

FindTargetDeploymentVersionAndRevisionNumberForWorkflowID returns the deployment version and revision number (if applicable) for the particular workflow ID based on the versioning info of the task queue. Nil means unversioned.

func FormatPinnedVersionNotInTaskQueueError 

func FormatPinnedVersionNotInTaskQueueError(deploymentName, buildID, taskQueue string, taskQueueType enumspb.TaskQueueType) string

FormatPinnedVersionNotInTaskQueueError formats the error message when a pinned version is not present in a task queue.

func GetOverrideOneTimeTargetVersion 

func GetOverrideOneTimeTargetVersion(override *workflowpb.VersioningOverride) *deploymentpb.WorkerDeploymentVersion

func GetOverridePinnedVersion 

func GetOverridePinnedVersion(override *workflowpb.VersioningOverride) *deploymentpb.WorkerDeploymentVersion

func GetOverrideTargetDeploymentVersion 

func GetOverrideTargetDeploymentVersion(override *workflowpb.VersioningOverride) *deploymentpb.WorkerDeploymentVersion

func HasDeploymentVersion 

func HasDeploymentVersion(deployments *persistencespb.DeploymentData, v *deploymentspb.WorkerDeploymentVersion) bool

func IsPinnedVersionNotInTaskQueueError 

func IsPinnedVersionNotInTaskQueueError(err error) bool

func IsUnversionedOrAssignedBuildIdSearchAttribute 

func IsUnversionedOrAssignedBuildIdSearchAttribute(buildId string) bool

IsUnversionedOrAssignedBuildIdSearchAttribute returns the value is "unversioned" or "assigned:<bld>"

func MakeBuildIdDirective 

func MakeBuildIdDirective(buildId string) *taskqueuespb.TaskVersionDirective

func MakeDirectiveForWorkflowTask 

func MakeDirectiveForWorkflowTask(
	inheritedBuildId string,
	assignedBuildId string,
	stamp *commonpb.WorkerVersionStamp,
	hasCompletedWorkflowTask bool,
	behavior enumspb.VersioningBehavior,
	deployment *deploymentpb.Deployment,
	revisionNumber int64,
	useRampingVersion bool,
) *taskqueuespb.TaskVersionDirective

MakeDirectiveForWorkflowTask returns a versioning directive based on the following parameters: - inheritedBuildId: build ID inherited from a past/previous wf execution (for Child WF or CaN) - assignedBuildId: the build ID to which the WF is currently assigned (i.e. mutable state's AssginedBuildId) - stamp: the latest versioning stamp of the execution (only needed for old versioning) - hasCompletedWorkflowTask: if the wf has completed any WFT - behavior: workflow's effective behavior - deployment: workflow's effective deployment

func MakeUseAssignmentRulesDirective 

func MakeUseAssignmentRulesDirective() *taskqueuespb.TaskVersionDirective

func OverrideIsPinned 

func OverrideIsPinned(override *workflowpb.VersioningOverride) bool

func PickFinalCurrentAndRamping 

func PickFinalCurrentAndRamping(
	current *deploymentspb.DeploymentVersionData,
	ramping *deploymentspb.DeploymentVersionData,
	currentVersionRoutingConfig *deploymentpb.RoutingConfig,
	rampingVersionRoutingConfig *deploymentpb.RoutingConfig,
) (
	finalCurrent *deploymentspb.WorkerDeploymentVersion,
	finalCurrentRev int64,
	finalCurrentUpdateTime time.Time,
	finalRamping *deploymentspb.WorkerDeploymentVersion,
	isRamping bool,
	finalRampPercentage float32,
	finalRampingRev int64,
	finalRampingUpdateTime time.Time,
)

PickFinalCurrentAndRamping determines the effective "current" and "ramping" deployment versions by comparing timestamps from the legacy deployment data (old format) and the RoutingConfig (new format). It returns: - final current deployment version and its revision number (0 for old format) - final ramping deployment version, its revision number (0 for old format), and ramp percentage

func PinnedBuildIdSearchAttribute 

func PinnedBuildIdSearchAttribute(version string) string

PinnedBuildIdSearchAttribute creates the pinned search attribute for the BuildIds list, used as a visibility optimization. For pinned workflows using WorkerDeployment APIs (ms.GetEffectiveVersioningBehavior() == PINNED && ms.executionInfo.VersioningInfo.Version != ""), this will be `pinned:<version>`. The version used will be the override version if set, or the versioningInfo.Version.

If deprecated Deployment-based APIs are in use and the workflow is pinned, `pinned:<deployment_series_name>:<deployment_build_id>` will. The values used will be the override deployment_series and build_id if set, or versioningInfo.Deployment.

If the workflow becomes unpinned or unversioned, this entry will be removed from that list.

func ShouldSkipReactivation 

func ShouldSkipReactivation(
	deployments *persistencespb.DeploymentData,
	deploymentName string,
	buildID string,
) (bool, int64)

ShouldSkipReactivation reports whether a reactivation signal to the given version would be redundant. Returns true when the version's status is CURRENT, RAMPING, or DRAINING. Returns false for DRAINED and INACTIVE (the two statuses the reactivation handler in version_workflow.go acts on) and when the version is not present in the deployment data. (UNSPECIFIED also yields false; in practice the deployment workflow sets a status at construction so this branch should not trigger.)

The returned revisionNumber is the version's revision as tracked in the new deployment data format (WorkerDeploymentVersionData.revision_number). It is 0 for the legacy DeploymentVersionData format (which does not carry a revision number) and when the version is not found at all.

func StampFromBuildId 

func StampFromBuildId(buildId string) *commonpb.WorkerVersionStamp

func StampFromCapabilities 

func StampFromCapabilities(capabilities *commonpb.WorkerVersionCapabilities, options *deploymentpb.WorkerDeploymentOptions) *commonpb.WorkerVersionStamp

func StampIfUsingVersioning 

func StampIfUsingVersioning(stamp *commonpb.WorkerVersionStamp) *commonpb.WorkerVersionStamp

StampIfUsingVersioning returns the given WorkerVersionStamp if it is using versioning, otherwise returns nil.

func UnversionedBuildIdSearchAttribute 

func UnversionedBuildIdSearchAttribute(buildId string) string

UnversionedBuildIdSearchAttribute returns the search attribute value for an unversioned build ID

func ValidateDeployment 

func ValidateDeployment(deployment *deploymentpb.Deployment) error

ValidateDeployment returns error if the deployment is nil or it has empty build ID or deployment name.

func ValidateDeploymentVersion 

func ValidateDeploymentVersion(version *deploymentspb.WorkerDeploymentVersion, maxIDLengthLimit int) error

ValidateDeploymentVersion returns error if the deployment version is not a valid entity.

func ValidateDeploymentVersionFields 

func ValidateDeploymentVersionFields(fieldName string, field string, maxIDLengthLimit int) error

ValidateDeploymentVersionFields is a helper that verifies if the fields within a Worker Deployment Version are valid

func ValidateDeploymentVersionStringV31 

func ValidateDeploymentVersionStringV31(version string) (*deploymentspb.WorkerDeploymentVersion, error)

ValidateDeploymentVersionStringV31 returns error if the deployment version is nil or it has empty version or deployment name.

func ValidateTaskVersionDirective 

func ValidateTaskVersionDirective(
	directive *taskqueuespb.TaskVersionDirective,
	wfBehavior enumspb.VersioningBehavior,
	wfDeployment *deploymentpb.Deployment,
	scheduledDeployment *deploymentpb.Deployment,
) error

func ValidateVersioningOverrideAndGetReactivationEligibility 

func ValidateVersioningOverrideAndGetReactivationEligibility(ctx context.Context,
	override *workflowpb.VersioningOverride,
	matchingClient resource.MatchingClient,
	versionCache VersionMembershipAndReactivationStatusCache,
	tq string,
	tqType enumspb.TaskQueueType,
	namespaceID string) (shouldSkipReactivation bool, revisionNumber int64, err error)

func ValidateVersioningOverrideStructure 

func ValidateVersioningOverrideStructure(override *workflowpb.VersioningOverride) error

func VersionStampToBuildIdSearchAttribute 

func VersionStampToBuildIdSearchAttribute(stamp *commonpb.WorkerVersionStamp) string

VersionStampToBuildIdSearchAttribute returns the search attribute value for a version stamp

func VersionedBuildIdSearchAttribute 

func VersionedBuildIdSearchAttribute(buildId string) string

VersionedBuildIdSearchAttribute returns the search attribute value for a versioned build ID

func WorkerDeploymentVersionFromStringV31 

func WorkerDeploymentVersionFromStringV31(s string) (*deploymentspb.WorkerDeploymentVersion, error)

func WorkerDeploymentVersionFromStringV32 

func WorkerDeploymentVersionFromStringV32(s string) (*deploymentspb.WorkerDeploymentVersion, error)

func WorkerDeploymentVersionToStringV31 

func WorkerDeploymentVersionToStringV31(v *deploymentspb.WorkerDeploymentVersion) string

func WorkerDeploymentVersionToStringV32 

func WorkerDeploymentVersionToStringV32(v *deploymentspb.WorkerDeploymentVersion) string

func WorkflowsExistForBuildId 

func WorkflowsExistForBuildId(ctx context.Context, visibilityManager manager.VisibilityManager, ns *namespace.Namespace, taskQueue, buildId string) (bool, error)

Types

type IsWFTaskQueueInVersionDetector 

type IsWFTaskQueueInVersionDetector = func(ctx context.Context, namespaceID, tq string, version *deploymentpb.WorkerDeploymentVersion) (bool, error)

func GetIsWFTaskQueueInVersionDetector 

func GetIsWFTaskQueueInVersionDetector(matchingClient resource.MatchingClient, versionCache VersionMembershipAndReactivationStatusCache) IsWFTaskQueueInVersionDetector

type RoutingInfo 

type RoutingInfo struct {
	Current               *deploymentspb.WorkerDeploymentVersion
	CurrentRevisionNumber int64
	Ramping               *deploymentspb.WorkerDeploymentVersion
	RampPercentage        float32
	RampingRevisionNumber int64
}

RoutingInfoCache is used to cache results of GetTaskQueueUserData calls followed by CalculateTaskQueueVersioningInfo computation.

Implementations are expected to be safe for concurrent use.

type RoutingInfoCache 

type RoutingInfoCache interface {
	// Get returns the cached routing info. ok=false means there was no cached value.
	Get(
		namespaceID string,
		taskQueue string,
		taskQueueType enumspb.TaskQueueType,
	) (RoutingInfo, bool)

	Put(
		namespaceID string,
		taskQueue string,
		taskQueueType enumspb.TaskQueueType,
		current *deploymentspb.WorkerDeploymentVersion,
		currentRevisionNumber int64,
		ramping *deploymentspb.WorkerDeploymentVersion,
		rampPercentage float32,
		rampingRevisionNumber int64,
	)
}

RoutingInfoCache is used to cache results of GetTaskQueueUserData calls followed by CalculateTaskQueueVersioningInfo computation.

Implementations are expected to be safe for concurrent use.

func NewRoutingInfoCache 

func NewRoutingInfoCache(c cache.Cache, metricsHandler metrics.Handler) RoutingInfoCache

NewRoutingInfoCache wraps the provided cache with a typed API and metrics.

type RoutingInfoCacheImpl 

type RoutingInfoCacheImpl struct {
	cache.Cache
	// contains filtered or unexported fields
}

RoutingInfoCache is used to cache results of GetTaskQueueUserData calls followed by CalculateTaskQueueVersioningInfo computation.

Implementations are expected to be safe for concurrent use.

func (*RoutingInfoCacheImpl) Get 

func (c *RoutingInfoCacheImpl) Get(
	namespaceID string,
	taskQueue string,
	taskQueueType enumspb.TaskQueueType,
) (RoutingInfo, bool)

func (*RoutingInfoCacheImpl) Put 

func (c *RoutingInfoCacheImpl) Put(
	namespaceID string,
	taskQueue string,
	taskQueueType enumspb.TaskQueueType,
	current *deploymentspb.WorkerDeploymentVersion,
	currentRevisionNumber int64,
	ramping *deploymentspb.WorkerDeploymentVersion,
	rampPercentage float32,
	rampingRevisionNumber int64,
)

type VersionMembershipAndReactivationStatusCache 

type VersionMembershipAndReactivationStatusCache interface {
	Get(
		namespaceID string,
		taskQueue string,
		taskQueueType enumspb.TaskQueueType,
		deploymentName string,
		buildID string,
	) (isMember bool, shouldSkipReactivation bool, revisionNumber int64, ok bool)

	Put(
		namespaceID string,
		taskQueue string,
		taskQueueType enumspb.TaskQueueType,
		deploymentName string,
		buildID string,
		isMember bool,
		shouldSkipReactivation bool,
		revisionNumber int64,
	)
}

VersionMembershipAndReactivationStatusCache caches results of Matching's CheckTaskQueueVersionMembership calls. It stores three pieces of information per version:

  • isMember: whether the task queue exists in the version (used for pinned override validation).
  • shouldSkipReactivation: true when the version's current status is CURRENT, RAMPING, or DRAINING, in which case the caller skips the reactivation signal. The zero value (false) is the safe default and covers unknown / not-found / old-matching cases.
  • revisionNumber: the version's current revision per matching's view. Used as part of the reactivation signal's RequestId so that all history pods targeting the same version at the same revision compose the same dedup key. Zero means unknown (old matching server or legacy DeploymentVersionData format with no revision_number field).

Implementations are expected to be safe for concurrent use.

func NewVersionMembershipAndReactivationStatusCache 

func NewVersionMembershipAndReactivationStatusCache(c cache.Cache, metricsHandler metrics.Handler) VersionMembershipAndReactivationStatusCache

NewVersionMembershipAndReactivationStatusCache wraps the provided cache with a typed API and metrics.

type VersionMembershipAndReactivationStatusCacheImpl 

type VersionMembershipAndReactivationStatusCacheImpl struct {
	cache.Cache
	// contains filtered or unexported fields
}

VersionMembershipAndReactivationStatusCache caches results of Matching's CheckTaskQueueVersionMembership calls. It stores three pieces of information per version:

  • isMember: whether the task queue exists in the version (used for pinned override validation).
  • shouldSkipReactivation: true when the version's current status is CURRENT, RAMPING, or DRAINING, in which case the caller skips the reactivation signal. The zero value (false) is the safe default and covers unknown / not-found / old-matching cases.
  • revisionNumber: the version's current revision per matching's view. Used as part of the reactivation signal's RequestId so that all history pods targeting the same version at the same revision compose the same dedup key. Zero means unknown (old matching server or legacy DeploymentVersionData format with no revision_number field).

Implementations are expected to be safe for concurrent use.

func (*VersionMembershipAndReactivationStatusCacheImpl) Get 

func (c *VersionMembershipAndReactivationStatusCacheImpl) Get(
	namespaceID string,
	taskQueue string,
	taskQueueType enumspb.TaskQueueType,
	deploymentName string,
	buildID string,
) (isMember bool, shouldSkipReactivation bool, revisionNumber int64, ok bool)

func (*VersionMembershipAndReactivationStatusCacheImpl) Put 

func (c *VersionMembershipAndReactivationStatusCacheImpl) Put(
	namespaceID string,
	taskQueue string,
	taskQueueType enumspb.TaskQueueType,
	deploymentName string,
	buildID string,
	isMember bool,
	shouldSkipReactivation bool,
	revisionNumber int64,
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.