Add priority annotations

Author: Sushisource

temporalio/lib/temporalio/activity/info.rb

@@ -10,6 +10,7 @@ module Activity
       :heartbeat_details,
       :heartbeat_timeout,
       :local?,
+      :priority,
       :schedule_to_close_timeout,
       :scheduled_time,
       :start_to_close_timeout,
@@ -38,6 +39,8 @@ module Activity
     #   @return [Float, nil] Heartbeat timeout set by the caller.
     # @!attribute local?
     #   @return [Boolean] Whether the activity is a local activity or not.
+    # @!attribute priority
+    #   @return [Priority, nil] The priority of this activity.
     # @!attribute schedule_to_close_timeout
     #   @return [Float, nil] Schedule to close timeout set by the caller.
     # @!attribute scheduled_time

temporalio/lib/temporalio/client.rb

@@ -19,6 +19,7 @@
 require 'temporalio/converters'
 require 'temporalio/error'
 require 'temporalio/internal/client/implementation'
+require 'temporalio/priority'
 require 'temporalio/retry_policy'
 require 'temporalio/runtime'
 require 'temporalio/search_attributes'
@@ -217,6 +218,9 @@ def operator_service
     #   with `cron_schedule`.
     # @param request_eager_start [Boolean] Potentially reduce the latency to start this workflow by encouraging the
     #   server to start it on a local worker running with this same client. This is currently experimental.
+    # @param versioning_override [VersioningOverride, nil] Override the version of the workflow.
+    #   This is currently experimental.
+    # @param priority [Priority] Priority of the workflow. This is currently experimental.
     # @param rpc_options [RPCOptions, nil] Advanced RPC options.
     #
     # @return [WorkflowHandle] A workflow handle to the started workflow.
@@ -241,6 +245,7 @@ def start_workflow(
       start_delay: nil,
       request_eager_start: false,
       versioning_override: nil,
+      priority: Priority.default,
       rpc_options: nil
     )
       @impl.start_workflow(Interceptor::StartWorkflowInput.new(
@@ -263,6 +268,7 @@ def start_workflow(
                              request_eager_start:,
                              headers: {},
                              versioning_override:,
+                             priority:,
                              rpc_options:
                            ))
     end
@@ -295,6 +301,9 @@ def start_workflow(
     #   with `cron_schedule`.
     # @param request_eager_start [Boolean] Potentially reduce the latency to start this workflow by encouraging the
     #   server to start it on a local worker running with this same client. This is currently experimental.
+    # @param versioning_override [VersioningOverride, nil] Override the version of the workflow.
+    #   This is currently experimental.
+    # @param priority [Priority] Priority for the workflow. This is currently experimental.
     # @param rpc_options [RPCOptions, nil] Advanced RPC options.
     #
     # @return [Object] Successful result of the workflow.
@@ -320,6 +329,7 @@ def execute_workflow(
       start_delay: nil,
       request_eager_start: false,
       versioning_override: nil,
+      priority: Priority.default,
       follow_runs: true,
       rpc_options: nil
     )
@@ -342,6 +352,7 @@ def execute_workflow(
         start_delay:,
         request_eager_start:,
         versioning_override:,
+        priority:,
         rpc_options:
       )
       follow_runs ? handle.result : handle

temporalio/lib/temporalio/client/interceptor.rb

@@ -37,6 +37,7 @@ def intercept_client(next_interceptor)
         :start_delay,
         :request_eager_start,
         :headers,
+        :priority,
         :versioning_override,
         :rpc_options
       )

temporalio/lib/temporalio/internal/client/implementation.rb

@@ -70,6 +70,7 @@ def start_workflow(input)
               input.static_summary, input.static_details, @client.data_converter
             ),
             header: ProtoUtils.headers_to_proto(input.headers, @client.data_converter),
+            priority: input.priority&._to_proto,
             versioning_override: input.versioning_override&._to_proto
           )
 

temporalio/lib/temporalio/internal/worker/activity_worker.rb

@@ -174,6 +174,7 @@ def execute_activity(task_token, defn, start)
             ),
             heartbeat_timeout: Internal::ProtoUtils.duration_to_seconds(start.heartbeat_timeout),
             local?: start.is_local,
+            priority: Priority._from_proto(start.priority),
             schedule_to_close_timeout: Internal::ProtoUtils.duration_to_seconds(start.schedule_to_close_timeout),
             scheduled_time: Internal::ProtoUtils.timestamp_to_time(start.scheduled_time) || raise, # Never nil
             start_to_close_timeout: Internal::ProtoUtils.duration_to_seconds(start.start_to_close_timeout),

temporalio/lib/temporalio/internal/worker/workflow_instance.rb

@@ -140,6 +140,7 @@ def initialize(details)
                           workflow_id: @init_job.parent_workflow_info.workflow_id
                         )
                       end,
+              priority: Priority._from_proto(@init_job.priority),
               retry_policy: (RetryPolicy._from_proto(@init_job.retry_policy) if @init_job.retry_policy),
               root: if @init_job.root_workflow
                       Workflow::Info::RootInfo.new(

temporalio/lib/temporalio/internal/worker/workflow_instance/context.rb

@@ -75,7 +75,8 @@ def execute_activity(
             cancellation:,
             cancellation_type:,
             activity_id:,
-            disable_eager_execution:
+            disable_eager_execution:,
+            priority:
           )
             activity = case activity
                        when Class
@@ -102,6 +103,7 @@ def execute_activity(
                 cancellation_type:,
                 activity_id:,
                 disable_eager_execution: disable_eager_execution || @instance.disable_eager_activity_execution,
+                priority:,
                 headers: {}
               )
             )
@@ -249,7 +251,8 @@ def start_child_workflow(
             retry_policy:,
             cron_schedule:,
             memo:,
-            search_attributes:
+            search_attributes:,
+            priority:
           )
             @outbound.start_child_workflow(
               Temporalio::Worker::Interceptor::Workflow::StartChildWorkflowInput.new(
@@ -270,6 +273,7 @@ def start_child_workflow(
                 cron_schedule:,
                 memo:,
                 search_attributes:,
+                priority:,
                 headers: {}
               )
             )

temporalio/lib/temporalio/internal/worker/workflow_instance/outbound_implementation.rb

@@ -73,7 +73,8 @@ def execute_activity(input)
                     heartbeat_timeout: ProtoUtils.seconds_to_duration(input.heartbeat_timeout),
                     retry_policy: input.retry_policy&._to_proto,
                     cancellation_type: input.cancellation_type,
-                    do_not_eagerly_execute: input.disable_eager_execution
+                    do_not_eagerly_execute: input.disable_eager_execution,
+                    priority: input.priority&._to_proto
                   ),
                   user_metadata: ProtoUtils.to_user_metadata(input.summary, nil, @instance.payload_converter)
                 )
@@ -337,7 +338,8 @@ def start_child_workflow(input)
                   headers: ProtoUtils.headers_to_proto_hash(input.headers, @instance.payload_converter),
                   memo: ProtoUtils.memo_to_proto_hash(input.memo, @instance.payload_converter),
                   search_attributes: input.search_attributes&._to_proto_hash,
-                  cancellation_type: input.cancellation_type
+                  cancellation_type: input.cancellation_type,
+                  priority: input.priority&._to_proto
                 ),
                 user_metadata: ProtoUtils.to_user_metadata(
                   input.static_summary, input.static_details, @instance.payload_converter

temporalio/lib/temporalio/priority.rb

@@ -0,0 +1,64 @@
+# frozen_string_literal: true
+
+require 'temporalio/api'
+
+module Temporalio
+  # Priority contains metadata that controls relative ordering of task processing when tasks are
+  # backlogged in a queue. Initially, Priority will be used in activity and workflow task
+  # queues, which are typically where backlogs exist. Priority is (for now) attached to
+  # workflows and activities. Activities and child workflows inherit Priority from the workflow
+  # that created them, but may override fields when they are started or modified. For each field
+  # of a Priority on an activity/workflow, 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 (documented on the field).
+  #
+  # The overall semantics of Priority are:
+  # 1. First, consider "priority_key": lower number goes first.
+  # (more will be added here later).
+  class Priority
+    # The priority key, which 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 priority is (min+max)/2. With the default max of 5 and min of 1, that comes
+    # out to 3.
+    #
+    # @return [Integer, nil] The priority key
+    attr_reader :priority_key
+
+    # @!visibility private
+    def self._from_proto(priority)
+      return default if priority.nil?
+
+      new(priority_key: priority.priority_key.zero? ? nil : priority.priority_key)
+    end
+
+    # The default priority instance.
+    #
+    # @return [Priority] The default priority
+    def self.default
+      @default ||= new
+    end
+
+    # Create a new Priority instance
+    #
+    # @param priority_key [Integer, nil] The priority key
+    def initialize(priority_key: nil)
+      @priority_key = priority_key
+    end
+
+    # @!visibility private
+    def _to_proto
+      return nil if priority_key.nil?
+
+      Temporalio::Api::Common::V1::Priority.new(priority_key: priority_key || 0)
+    end
+
+    # @return [Boolean] True if this priority is empty/default
+    def empty?
+      priority_key.nil?
+    end
+  end
+end

temporalio/lib/temporalio/testing/activity_environment.rb

@@ -24,6 +24,7 @@ def self.default_info
           heartbeat_details: [],
           heartbeat_timeout: nil,
           local?: false,
+          priority: Temporalio::Priority.default,
           schedule_to_close_timeout: 1.0,
           scheduled_time: Time.at(0),
           start_to_close_timeout: 1.0,