Priority field passthrough (#892)

Author: Sushisource

.cargo/config.toml

@@ -1,3 +1,8 @@
+[env]
+# This temporarily overrides the version of the CLI used for integration tests, locally and in CI
+# TODO: Comment out once priority is released
+CLI_VERSION_OVERRIDE = "v1.3.1-priority.0"
+
 [alias]
 integ-test = ["run", "--package", "temporal-sdk-core", "--example", "integ_runner", "--"]
 lint = ["clippy", "--workspace", "--examples", "--all-features",

client/src/lib.rs

@@ -1088,6 +1088,51 @@ pub struct WorkflowOptions {
     /// Callbacks that will be invoked upon workflow completion. For, ex, completing nexus
     /// operations.
     pub completion_callbacks: Vec<common::v1::Callback>,
+
+    /// Priority for the workflow
+    pub priority: Option<Priority>,
+}
+
+/// 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.
+/// Other queues in the server (such as transfer and timer queues) and rate
+/// limiting decisions do not use Priority, but may in the future.
+///
+/// Priority is 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 below).
+///
+/// Despite being named "Priority", this message will also contains fields that
+/// control "fairness" mechanisms.
+///
+/// The overall semantics of Priority are:
+/// (more will be added here later)
+/// 1. First, consider "priority_key": lower number goes first.
+#[derive(Debug, Clone, Default)]
+pub struct Priority {
+    /// Priority key 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.
+    pub priority_key: u32,
+}
+
+impl From<Priority> for common::v1::Priority {
+    fn from(priority: Priority) -> Self {
+        common::v1::Priority {
+            priority_key: priority.priority_key as i32,
+        }
+    }
 }
 
 #[async_trait::async_trait]
@@ -1140,6 +1185,7 @@ where
                 retry_policy: options.retry_policy,
                 links: options.links,
                 completion_callbacks: options.completion_callbacks,
+                priority: options.priority.map(Into::into),
                 ..Default::default()
             })
             .await?

client/src/raw.rs

@@ -1178,27 +1178,27 @@ proxier! {
         }
     );
     (
-        update_activity_options_by_id,
-        UpdateActivityOptionsByIdRequest,
-        UpdateActivityOptionsByIdResponse,
+        update_activity_options,
+        UpdateActivityOptionsRequest,
+        UpdateActivityOptionsResponse,
         |r| {
             let labels = namespaced_request!(r);
             r.extensions_mut().insert(labels);
         }
     );
     (
-        pause_activity_by_id,
-        PauseActivityByIdRequest,
-        PauseActivityByIdResponse,
+        pause_activity,
+        PauseActivityRequest,
+        PauseActivityResponse,
         |r| {
             let labels = namespaced_request!(r);
             r.extensions_mut().insert(labels);
         }
     );
     (
-        unpause_activity_by_id,
-        UnpauseActivityByIdRequest,
-        UnpauseActivityByIdResponse,
+        unpause_activity,
+        UnpauseActivityRequest,
+        UnpauseActivityResponse,
         |r| {
             let labels = namespaced_request!(r);
             r.extensions_mut().insert(labels);
@@ -1214,9 +1214,81 @@ proxier! {
         }
     );
     (
-        reset_activity_by_id,
-        ResetActivityByIdRequest,
-        ResetActivityByIdResponse,
+        reset_activity,
+        ResetActivityRequest,
+        ResetActivityResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        delete_worker_deployment,
+        DeleteWorkerDeploymentRequest,
+        DeleteWorkerDeploymentResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        delete_worker_deployment_version,
+        DeleteWorkerDeploymentVersionRequest,
+        DeleteWorkerDeploymentVersionResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        describe_worker_deployment,
+        DescribeWorkerDeploymentRequest,
+        DescribeWorkerDeploymentResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        describe_worker_deployment_version,
+        DescribeWorkerDeploymentVersionRequest,
+        DescribeWorkerDeploymentVersionResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        list_worker_deployments,
+        ListWorkerDeploymentsRequest,
+        ListWorkerDeploymentsResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        set_worker_deployment_current_version,
+        SetWorkerDeploymentCurrentVersionRequest,
+        SetWorkerDeploymentCurrentVersionResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        set_worker_deployment_ramping_version,
+        SetWorkerDeploymentRampingVersionRequest,
+        SetWorkerDeploymentRampingVersionResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        update_worker_deployment_version_metadata,
+        UpdateWorkerDeploymentVersionMetadataRequest,
+        UpdateWorkerDeploymentVersionMetadataResponse,
         |r| {
             let labels = namespaced_request!(r);
             r.extensions_mut().insert(labels);

core/src/worker/client.rs

@@ -180,12 +180,14 @@ impl WorkerClient for WorkerClientBag {
         &self,
         task_queue: TaskQueue,
     ) -> Result<PollWorkflowTaskQueueResponse> {
+        #[allow(deprecated)] // want to list all fields explicitly
         let mut request = PollWorkflowTaskQueueRequest {
             namespace: self.namespace.clone(),
             task_queue: Some(task_queue),
             identity: self.identity.clone(),
             binary_checksum: self.binary_checksum(),
             worker_version_capabilities: self.worker_version_capabilities(),
+            deployment_options: None,
         }
         .into_request();
         request.extensions_mut().insert(IsWorkerTaskLongPoll);
@@ -202,6 +204,7 @@ impl WorkerClient for WorkerClientBag {
         task_queue: String,
         max_tasks_per_sec: Option<f64>,
     ) -> Result<PollActivityTaskQueueResponse> {
+        #[allow(deprecated)] // want to list all fields explicitly
         let mut request = PollActivityTaskQueueRequest {
             namespace: self.namespace.clone(),
             task_queue: Some(TaskQueue {
@@ -214,6 +217,7 @@ impl WorkerClient for WorkerClientBag {
                 max_tasks_per_second: Some(tps),
             }),
             worker_version_capabilities: self.worker_version_capabilities(),
+            deployment_options: None,
         }
         .into_request();
         request.extensions_mut().insert(IsWorkerTaskLongPoll);
@@ -226,6 +230,7 @@ impl WorkerClient for WorkerClientBag {
     }
 
     async fn poll_nexus_task(&self, task_queue: String) -> Result<PollNexusTaskQueueResponse> {
+        #[allow(deprecated)] // want to list all fields explicitly
         let mut request = PollNexusTaskQueueRequest {
             namespace: self.namespace.clone(),
             task_queue: Some(TaskQueue {
@@ -235,6 +240,7 @@ impl WorkerClient for WorkerClientBag {
             }),
             identity: self.identity.clone(),
             worker_version_capabilities: self.worker_version_capabilities(),
+            deployment_options: None,
         }
         .into_request();
         request.extensions_mut().insert(IsWorkerTaskLongPoll);
@@ -287,6 +293,7 @@ impl WorkerClient for WorkerClientBag {
             // TODO: https://github.com/temporalio/sdk-core/issues/866
             deployment: None,
             versioning_behavior: 0,
+            deployment_options: None,
         };
         Ok(self
             .cloned_client()
@@ -312,6 +319,7 @@ impl WorkerClient for WorkerClientBag {
                     worker_version: self.worker_version_stamp(),
                     // TODO: https://github.com/temporalio/sdk-core/issues/866
                     deployment: None,
+                    deployment_options: None,
                 },
             )
             .await?
@@ -369,6 +377,7 @@ impl WorkerClient for WorkerClientBag {
                     worker_version: self.worker_version_stamp(),
                     // TODO: https://github.com/temporalio/sdk-core/issues/866
                     deployment: None,
+                    deployment_options: None,
                 },
             )
             .await?
@@ -394,6 +403,7 @@ impl WorkerClient for WorkerClientBag {
                     worker_version: self.worker_version_stamp(),
                     // TODO: https://github.com/temporalio/sdk-core/issues/866
                     deployment: None,
+                    deployment_options: None,
                 },
             )
             .await?
@@ -418,6 +428,7 @@ impl WorkerClient for WorkerClientBag {
             worker_version: self.worker_version_stamp(),
             // TODO: https://github.com/temporalio/sdk-core/issues/866
             deployment: None,
+            deployment_options: None,
         };
         Ok(self
             .cloned_client()

core/src/worker/workflow/machines/nexus_operation_state_machine.rs

@@ -534,6 +534,7 @@ impl NexusOperationMachine {
                     service: self.shared_state.service.clone(),
                     operation: self.shared_state.operation.clone(),
                     operation_id: "".to_string(),
+                    operation_token: "".to_string(),
                 },
             )),
             ..Default::default()

sdk-core-protos/protos/api_upstream/Makefile

@@ -1,4 +1,4 @@
-SHELL=/bin/bash -o pipefail
+SHELL=bash -o pipefail
 
 $(VERBOSE).SILENT:
 ############################# Main targets #############################
@@ -76,7 +76,7 @@ http-api-docs:
 ##### Plugins & tools #####
 grpc-install:
 	@printf $(COLOR) "Install/update protoc and plugins..."
-	@go install go.temporal.io/api/cmd/protogen@latest
+	@go install go.temporal.io/api/cmd/protogen@master
 	@go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
 	@go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
 	@go install github.com/grpc-ecosystem/grpc-gateway/v2/protoc-gen-grpc-gateway@latest