Add priority annotations (#1792)

stephanos · web-flow · commit 081958b079d6 · 2025-02-10T13:07:07.000-08:00
diff --git a/internal/activity.go b/internal/activity.go
@@ -62,6 +62,7 @@ type (
 		Deadline          time.Time     // Time of activity timeout
 		Attempt           int32         // Attempt starts from 1, and increased by 1 for every retry if retry policy is specified.
 		IsLocalActivity   bool          // true if it is a local activity
+		Priority          Priority
 	}
 
 	// RegisterActivityOptions consists of options for registering an activity.
@@ -165,6 +166,10 @@ type (
 		//
 		// NOTE: Experimental
 		Summary string
+
+		// Priority - Optional priority settings that control relative ordering of
+		// task processing when tasks are backed up in a queue.
+		Priority Priority
 	}
 
 	// LocalActivityOptions stores local activity specific parameters that will be stored inside of a context.
@@ -320,6 +325,7 @@ func WithActivityTask(
 		taskQueue:        taskQueue,
 		dataConverter:    dataConverter,
 		attempt:          task.GetAttempt(),
+		priority:         task.GetPriority(),
 		heartbeatDetails: task.HeartbeatDetails,
 		workflowType: &WorkflowType{
 			Name: task.WorkflowType.GetName(),
diff --git a/internal/client.go b/internal/client.go
@@ -774,6 +774,10 @@ type (
 		// NOTE: Experimental
 		VersioningOverride VersioningOverride
 
+		// Priority - Optional priority settings that control relative ordering of
+		// task processing when tasks are backed up in a queue.
+		Priority Priority
+
 		// request ID. Only settable by the SDK - e.g. [temporalnexus.workflowRunOperation].
 		requestID string
 		// workflow completion callback. Only settable by the SDK - e.g. [temporalnexus.workflowRunOperation].
@@ -835,6 +839,33 @@ type (
 		NonRetryableErrorTypes []string
 	}
 
+	// Priority contains metadata that controls the relative ordering of task processing
+	// when tasks are backed up in a queue. The affected queues depend on the
+	// server version.
+	//
+	// Priority is attached to workflows and activities. By default, activities
+	// and child workflows inherit Priority from the workflow that created them, but may
+	// override fields when an activity is started or modified.
+	//
+	// For all fields, the field not present or equal to zero/empty string means to
+	// inherit the value from the calling workflow, or if there is no calling
+	// workflow, then use the default value.
+	//
+	// Exposed as: [go.temporal.io/sdk/temporal.Priority]
+	Priority struct {
+		// PriorityKey is a positive integer from 1 to n, where smaller integers
+		// correspond to higher priorities (tasks run sooner). In general, tasks in
+		// a queue should be processed in close to priority order, although small
+		// deviations are possible.
+		//
+		// The maximum priority value (minimum priority) is determined by server
+		// configuration, and defaults to 5.
+		//
+		// The default value when unset or 0 is calculated by (min+max)/2. With the
+		// default max of 5, and min of 1, that comes out to 3.
+		PriorityKey int
+	}
+
 	// NamespaceClient is the client for managing operations on the namespace.
 	// CLI, tools, ... can use this layer to manager operations on namespace.
 	NamespaceClient interface {
diff --git a/internal/internal_activity.go b/internal/internal_activity.go
@@ -74,6 +74,7 @@ type (
 		DisableEagerExecution  bool
 		VersioningIntent       VersioningIntent
 		Summary                string
+		Priority               *commonpb.Priority
 	}
 
 	// ExecuteLocalActivityOptions options for executing a local activity
@@ -146,6 +147,7 @@ type (
 		workflowNamespace  string
 		workerStopChannel  <-chan struct{}
 		contextPropagators []ContextPropagator
+		priority           *commonpb.Priority
 	}
 
 	// context.WithValue need this type instead of basic type string to avoid lint error
@@ -382,6 +384,7 @@ func (a *activityEnvironmentInterceptor) GetInfo(ctx context.Context) ActivityIn
 		WorkflowType:      a.env.workflowType,
 		WorkflowNamespace: a.env.workflowNamespace,
 		IsLocalActivity:   a.env.isLocalActivity,
+		Priority:          convertFromPBPriority(a.env.priority),
 	}
 }
 
diff --git a/internal/internal_event_handlers.go b/internal/internal_event_handlers.go
@@ -593,6 +593,7 @@ func (wc *workflowEnvironmentImpl) ExecuteChildWorkflow(
 	attributes.WorkflowIdReusePolicy = params.WorkflowIDReusePolicy
 	attributes.ParentClosePolicy = params.ParentClosePolicy
 	attributes.RetryPolicy = params.RetryPolicy
+	attributes.Priority = params.Priority
 	attributes.Header = params.Header
 	attributes.Memo = memo
 	attributes.SearchAttributes = searchAttr
@@ -758,6 +759,7 @@ func (wc *workflowEnvironmentImpl) ExecuteActivity(parameters ExecuteActivityPar
 	scheduleTaskAttr.RequestEagerExecution = !parameters.DisableEagerExecution
 	scheduleTaskAttr.UseWorkflowBuildId = determineInheritBuildIdFlagForCommand(
 		parameters.VersioningIntent, wc.workflowInfo.TaskQueueName, parameters.TaskQueueName)
+	scheduleTaskAttr.Priority = parameters.Priority
 
 	startMetadata, err := buildUserMetadata(parameters.Summary, "", wc.dataConverter)
 	if err != nil {
diff --git a/internal/internal_schedule_client.go b/internal/internal_schedule_client.go
@@ -661,6 +661,7 @@ func convertToPBScheduleAction(
 					Header:                   header,
 					UserMetadata:             userMetadata,
 					VersioningOverride:       versioningOverrideToProto(action.VersioningOverride),
+					Priority:                 convertToPBPriority(action.Priority),
 				},
 			},
 		}, nil
diff --git a/internal/internal_task_handlers.go b/internal/internal_task_handlers.go
@@ -744,6 +744,7 @@ func (wth *workflowTaskHandlerImpl) createWorkflowContext(task *workflowservice.
 		Memo:                     attributes.Memo,
 		SearchAttributes:         attributes.SearchAttributes,
 		RetryPolicy:              convertFromPBRetryPolicy(attributes.RetryPolicy),
+		Priority:                 convertFromPBPriority(attributes.Priority),
 	}
 
 	return newWorkflowExecutionContext(workflowInfo, wth), nil
diff --git a/internal/internal_workflow.go b/internal/internal_workflow.go
@@ -218,6 +218,7 @@ type (
 		WorkflowIDConflictPolicy enumspb.WorkflowIdConflictPolicy
 		DataConverter            converter.DataConverter
 		RetryPolicy              *commonpb.RetryPolicy
+		Priority                 *commonpb.Priority
 		CronSchedule             string
 		ContextPropagators       []ContextPropagator
 		Memo                     map[string]interface{}
diff --git a/internal/internal_workflow_client.go b/internal/internal_workflow_client.go
@@ -740,11 +740,12 @@ func (wc *WorkflowClient) DescribeWorkflowExecution(ctx context.Context, workflo
 
 // QueryWorkflow queries a given workflow execution
 // workflowID and queryType are required, other parameters are optional.
-//  - workflow ID of the workflow.
-//  - runID can be default(empty string). if empty string then it will pick the running execution of that workflow ID.
-//  - taskQueue can be default(empty string). If empty string then it will pick the taskQueue of the running execution of that workflow ID.
-//  - queryType is the type of the query.
-//  - args... are the optional query parameters.
+//   - workflow ID of the workflow.
+//   - runID can be default(empty string). if empty string then it will pick the running execution of that workflow ID.
+//   - taskQueue can be default(empty string). If empty string then it will pick the taskQueue of the running execution of that workflow ID.
+//   - queryType is the type of the query.
+//   - args... are the optional query parameters.
+//
 // The errors it can return:
 //   - serviceerror.InvalidArgument
 //   - serviceerror.Internal
@@ -943,8 +944,9 @@ func (wc *WorkflowClient) QueryWorkflowWithOptions(ctx context.Context, request
 
 // DescribeTaskQueue returns information about the target taskqueue, right now this API returns the
 // pollers which polled this taskqueue in last few minutes.
-//  - taskqueue name of taskqueue
-//  - taskqueueType type of taskqueue, can be workflow or activity
+//   - taskqueue name of taskqueue
+//   - taskqueueType type of taskqueue, can be workflow or activity
+//
 // The errors it can return:
 //   - serviceerror.InvalidArgument
 //   - serviceerror.Internal
@@ -1684,6 +1686,7 @@ func (w *workflowClientInterceptor) createStartWorkflowRequest(
 		CompletionCallbacks:      in.Options.callbacks,
 		Links:                    in.Options.links,
 		VersioningOverride:       versioningOverrideToProto(in.Options.VersioningOverride),
+		Priority:                 convertToPBPriority(in.Options.Priority),
 	}
 
 	startRequest.UserMetadata, err = buildUserMetadata(in.Options.StaticSummary, in.Options.StaticDetails, dataConverter)
@@ -2056,6 +2059,7 @@ func (w *workflowClientInterceptor) SignalWithStartWorkflow(
 		WorkflowIdConflictPolicy: in.Options.WorkflowIDConflictPolicy,
 		Header:                   header,
 		VersioningOverride:       versioningOverrideToProto(in.Options.VersioningOverride),
+		Priority:                 convertToPBPriority(in.Options.Priority),
 	}
 
 	if in.Options.StartDelay != 0 {
diff --git a/internal/internal_workflow_testsuite.go b/internal/internal_workflow_testsuite.go
@@ -2155,6 +2155,7 @@ func newTestActivityTask(workflowID, runID, workflowTypeName, namespace string,
 		},
 		WorkflowNamespace: namespace,
 		Header:            attr.GetHeader(),
+		Priority:          attr.Priority,
 	}
 	return task
 }
diff --git a/internal/nexus_operations.go b/internal/nexus_operations.go
@@ -241,6 +241,7 @@ func (t *testSuiteClientForNexusOperations) ExecuteWorkflow(ctx context.Context,
 				Memo:                     options.Memo,
 				CronSchedule:             options.CronSchedule,
 				RetryPolicy:              convertToPBRetryPolicy(options.RetryPolicy),
+				Priority:                 convertToPBPriority(options.Priority),
 			},
 		}, func(result *commonpb.Payloads, wfErr error) {
 			ncb := callback.GetNexus()
diff --git a/internal/schedule_client.go b/internal/schedule_client.go
@@ -318,6 +318,10 @@ type (
 		//
 		// NOTE: Experimental
 		StaticDetails string
+
+		// Priority - Optional priority settings that control relative ordering of
+		// task processing when tasks are backed up in a queue.
+		Priority Priority
 	}
 
 	// ScheduleOptions configure the parameters for creating a schedule.
diff --git a/internal/workflow.go b/internal/workflow.go
@@ -434,6 +434,10 @@ type (
 		//
 		// NOTE: Experimental
 		StaticDetails string
+
+		// Priority - Optional priority settings that control relative ordering of
+		// task processing when tasks are backed up in a queue.
+		Priority Priority
 	}
 
 	// RegisterWorkflowOptions consists of options for registering a workflow
@@ -1267,6 +1271,7 @@ type WorkflowInfo struct {
 	// Deprecated: use [Workflow.GetTypedSearchAttributes] instead.
 	SearchAttributes *commonpb.SearchAttributes // Value can be decoded using defaultDataConverter.
 	RetryPolicy      *RetryPolicy
+	Priority         Priority
 	// BinaryChecksum represents the value persisted by the last worker to complete a task in this workflow. It may be
 	// an explicitly set or implicitly derived binary checksum of the worker binary, or, if this worker has opted into
 	// build-id based versioning, is the explicitly set worker build id. If this is the first worker to operate on the
@@ -1746,6 +1751,7 @@ func WithChildWorkflowOptions(ctx Context, cwo ChildWorkflowOptions) Context {
 	wfOptions.VersioningIntent = cwo.VersioningIntent
 	wfOptions.StaticSummary = cwo.StaticSummary
 	wfOptions.StaticDetails = cwo.StaticDetails
+	wfOptions.Priority = convertToPBPriority(cwo.Priority)
 
 	return ctx1
 }
@@ -1768,6 +1774,7 @@ func GetChildWorkflowOptions(ctx Context) ChildWorkflowOptions {
 		WaitForCancellation:      opts.WaitForCancellation,
 		WorkflowIDReusePolicy:    opts.WorkflowIDReusePolicy,
 		RetryPolicy:              convertFromPBRetryPolicy(opts.RetryPolicy),
+		Priority:                 convertFromPBPriority(opts.Priority),
 		CronSchedule:             opts.CronSchedule,
 		Memo:                     opts.Memo,
 		SearchAttributes:         opts.SearchAttributes,
@@ -2338,6 +2345,7 @@ func WithActivityOptions(ctx Context, options ActivityOptions) Context {
 	eap.RetryPolicy = convertToPBRetryPolicy(options.RetryPolicy)
 	eap.DisableEagerExecution = options.DisableEagerExecution
 	eap.VersioningIntent = options.VersioningIntent
+	eap.Priority = convertToPBPriority(options.Priority)
 	eap.Summary = options.Summary
 	return ctx1
 }
@@ -2401,6 +2409,7 @@ func GetActivityOptions(ctx Context) ActivityOptions {
 		RetryPolicy:            convertFromPBRetryPolicy(opts.RetryPolicy),
 		DisableEagerExecution:  opts.DisableEagerExecution,
 		VersioningIntent:       opts.VersioningIntent,
+		Priority:               convertFromPBPriority(opts.Priority),
 		Summary:                opts.Summary,
 	}
 }
@@ -2513,6 +2522,32 @@ func convertFromPBRetryPolicy(retryPolicy *commonpb.RetryPolicy) *RetryPolicy {
 	return &p
 }
 
+func convertToPBPriority(priority Priority) *commonpb.Priority {
+	// If the priority only contains default values, return nil instead
+	// - since there's no need to send the default values to the server.
+	//
+	// Exposed as: [go.temporal.io/sdk/temporal.Priority]
+	var defaultPriority Priority
+	if priority == defaultPriority {
+		return nil
+	}
+
+	return &commonpb.Priority{
+		PriorityKey: int32(priority.PriorityKey),
+	}
+}
+
+func convertFromPBPriority(priority *commonpb.Priority) Priority {
+	// If the priority is nil, return the default value.
+	if priority == nil {
+		return Priority{}
+	}
+
+	return Priority{
+		PriorityKey: int(priority.PriorityKey),
+	}
+}
+
 // GetLastCompletionResultFromWorkflowInfo returns value of last completion result.
 func GetLastCompletionResultFromWorkflowInfo(info *WorkflowInfo) *commonpb.Payloads {
 	return info.lastCompletionResult
diff --git a/internal/workflow_test.go b/internal/workflow_test.go
@@ -63,6 +63,7 @@ func TestGetChildWorkflowOptions(t *testing.T) {
 		VersioningIntent:  VersioningIntentDefault,
 		StaticSummary:     "child workflow summary",
 		StaticDetails:     "child workflow details",
+		Priority:          newPriority(),
 	}
 
 	// Require test options to have non-zero value for each field. This ensures that we update tests (and the
@@ -85,6 +86,7 @@ func TestGetActivityOptions(t *testing.T) {
 		DisableEagerExecution:  true,
 		VersioningIntent:       VersioningIntentDefault,
 		Summary:                "activity summary",
+		Priority:               newPriority(),
 	}
 
 	assertNonZero(t, opts)
@@ -141,6 +143,12 @@ func newTestRetryPolicy() *RetryPolicy {
 	}
 }
 
+func newPriority() Priority {
+	return Priority{
+		PriorityKey: 1,
+	}
+}
+
 // assertNonZero checks that every top level value, struct field, and item in a slice is a non-zero value.
 func assertNonZero(t *testing.T, i interface{}) {
 	_assertNonZero(t, i, reflect.ValueOf(i).Type().Name())
diff --git a/temporal/priority.go b/temporal/priority.go
@@ -0,0 +1,28 @@
+// The MIT License
+//
+// Copyright (c) 2025 Temporal Technologies Inc.  All rights reserved.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package temporal
+
+import "go.temporal.io/sdk/internal"
+
+// Priority defines the priority for activity/workflow.
+type Priority = internal.Priority

Original file line number	Diff line number	Diff line change
`@@ -74,6 +74,7 @@ type (`
`74`	`74`	`DisableEagerExecution bool`
`75`	`75`	`VersioningIntent VersioningIntent`
`76`	`76`	`Summary string`
	`77`	`+ Priority *commonpb.Priority`
`77`	`78`	`}`
`78`	`79`
`79`	`80`	`// ExecuteLocalActivityOptions options for executing a local activity`
`@@ -146,6 +147,7 @@ type (`
`146`	`147`	`workflowNamespace string`
`147`	`148`	`workerStopChannel <-chan struct{}`
`148`	`149`	`contextPropagators []ContextPropagator`
	`150`	`+ priority *commonpb.Priority`
`149`	`151`	`}`
`150`	`152`
`151`	`153`	`// context.WithValue need this type instead of basic type string to avoid lint error`
`@@ -382,6 +384,7 @@ func (a *activityEnvironmentInterceptor) GetInfo(ctx context.Context) ActivityIn`
`382`	`384`	`WorkflowType: a.env.workflowType,`
`383`	`385`	`WorkflowNamespace: a.env.workflowNamespace,`
`384`	`386`	`IsLocalActivity: a.env.isLocalActivity,`
	`387`	`+ Priority: convertFromPBPriority(a.env.priority),`
`385`	`388`	`}`
`386`	`389`	`}`
`387`	`390`
Original file line number	Diff line number	Diff line change
`@@ -744,6 +744,7 @@ func (wth workflowTaskHandlerImpl) createWorkflowContext(task workflowservice.`
`744`	`744`	`Memo: attributes.Memo,`
`745`	`745`	`SearchAttributes: attributes.SearchAttributes,`
`746`	`746`	`RetryPolicy: convertFromPBRetryPolicy(attributes.RetryPolicy),`
	`747`	`+ Priority: convertFromPBPriority(attributes.Priority),`
`747`	`748`	`}`
`748`	`749`
`749`	`750`	`return newWorkflowExecutionContext(workflowInfo, wth), nil`
Original file line number	Diff line number	Diff line change
`@@ -2155,6 +2155,7 @@ func newTestActivityTask(workflowID, runID, workflowTypeName, namespace string,`
`2155`	`2155`	`},`
`2156`	`2156`	`WorkflowNamespace: namespace,`
`2157`	`2157`	`Header: attr.GetHeader(),`
	`2158`	`+ Priority: attr.Priority,`
`2158`	`2159`	`}`
`2159`	`2160`	`return task`
`2160`	`2161`	`}`