Fairness protos (#971)

Author: Sushisource

.cargo/config.toml

@@ -1,6 +1,6 @@
 [env]
 # This temporarily overrides the version of the CLI used for integration tests, locally and in CI
-#CLI_VERSION_OVERRIDE = "v1.3.1-priority.0"
+CLI_VERSION_OVERRIDE = "v1.4.1-cloud-v1-29-0-139-2.0"
 
 [alias]
 integ-test = ["test", "--features", "temporal-sdk-core-protos/serde_serialize", "--package", "temporal-sdk-core", "--test", "integ_runner", "--"]

client/src/lib.rs

@@ -1142,7 +1142,7 @@ pub struct WorkflowOptions {
 /// 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, PartialEq, Eq)]
+#[derive(Debug, Clone, Default, PartialEq)]
 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
@@ -1155,12 +1155,50 @@ pub struct Priority {
     /// 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,
+
+    /// Fairness key is a short string that's used as a key for a fairness
+    /// balancing mechanism. It may correspond to a tenant id, or to a fixed
+    /// string like "high" or "low". The default is the empty string.
+    ///
+    /// The fairness mechanism attempts to dispatch tasks for a given key in
+    /// proportion to its weight. For example, using a thousand distinct tenant
+    /// ids, each with a weight of 1.0 (the default) will result in each tenant
+    /// getting a roughly equal share of task dispatch throughput.
+    ///
+    /// (Note: this does not imply equal share of worker capacity! Fairness
+    /// decisions are made based on queue statistics, not
+    /// current worker load.)
+    ///
+    /// As another example, using keys "high" and "low" with weight 9.0 and 1.0
+    /// respectively will prefer dispatching "high" tasks over "low" tasks at a
+    /// 9:1 ratio, while allowing either key to use all worker capacity if the
+    /// other is not present.
+    ///
+    /// All fairness mechanisms, including rate limits, are best-effort and
+    /// probabilistic. The results may not match what a "perfect" algorithm with
+    /// infinite resources would produce. The more unique keys are used, the less
+    /// accurate the results will be.
+    ///
+    /// Fairness keys are limited to 64 bytes.
+    pub fairness_key: String,
+
+    /// Fairness weight for a task can come from multiple sources for
+    /// flexibility. From highest to lowest precedence:
+    /// 1. Weights for a small set of keys can be overridden in task queue
+    ///    configuration with an API.
+    /// 2. It can be attached to the workflow/activity in this field.
+    /// 3. The default weight of 1.0 will be used.
+    ///
+    /// Weight values are clamped by the server to the range [0.001, 1000].
+    pub fairness_weight: f32,
 }
 
 impl From<Priority> for common::v1::Priority {
     fn from(priority: Priority) -> Self {
         common::v1::Priority {
             priority_key: priority.priority_key as i32,
+            fairness_key: priority.fairness_key,
+            fairness_weight: priority.fairness_weight,
         }
     }
 }
@@ -1169,6 +1207,8 @@ impl From<common::v1::Priority> for Priority {
     fn from(priority: common::v1::Priority) -> Self {
         Self {
             priority_key: priority.priority_key as u32,
+            fairness_key: priority.fairness_key,
+            fairness_weight: priority.fairness_weight,
         }
     }
 }

client/src/raw.rs

@@ -1354,6 +1354,34 @@ proxier! {
             r.extensions_mut().insert(labels);
         }
     );
+    (
+        update_task_queue_config,
+        UpdateTaskQueueConfigRequest,
+        UpdateTaskQueueConfigResponse,
+        |r| {
+            let mut labels = namespaced_request!(r);
+            labels.task_q_str(r.get_ref().task_queue.clone());
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        fetch_worker_config,
+        FetchWorkerConfigRequest,
+        FetchWorkerConfigResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        update_worker_config,
+        UpdateWorkerConfigRequest,
+        UpdateWorkerConfigResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
 }
 
 proxier! {
@@ -1543,11 +1571,17 @@ mod tests {
             })
             .collect();
         let no_underscores: HashSet<_> = impl_list.iter().map(|x| x.replace('_', "")).collect();
+        let mut not_implemented = vec![];
         for method in methods {
             if !no_underscores.contains(&method.to_lowercase()) {
-                panic!("RPC method {method} is not implemented by raw client")
+                not_implemented.push(method);
             }
         }
+        if !not_implemented.is_empty() {
+            panic!(
+                "The following RPC methods are not implemented by raw client: {not_implemented:?}"
+            );
+        }
     }
     #[test]
     fn verify_all_workflow_service_methods_implemented() {

core-c-bridge/src/client.rs

@@ -1,12 +1,8 @@
-use crate::ByteArray;
-use crate::ByteArrayRef;
-use crate::CancellationToken;
-use crate::MetadataRef;
-use crate::UserDataHandle;
-use crate::runtime::Runtime;
-
-use std::str::FromStr;
-use std::time::Duration;
+use crate::{
+    ByteArray, ByteArrayRef, CancellationToken, MetadataRef, UserDataHandle, runtime::Runtime,
+};
+
+use std::{str::FromStr, time::Duration};
 use temporal_client::{
     ClientKeepAliveConfig, ClientOptions as CoreClientOptions, ClientOptionsBuilder,
     ClientTlsConfig, CloudService, ConfiguredClient, HealthService, HttpConnectProxyOptions,
@@ -326,6 +322,7 @@ async fn call_workflow_service(
         "DescribeWorkflowExecution" => rpc_call!(client, call, describe_workflow_execution),
         "DescribeWorkflowRule" => rpc_call!(client, call, describe_workflow_rule),
         "ExecuteMultiOperation" => rpc_call!(client, call, execute_multi_operation),
+        "FetchWorkerConfig" => rpc_call!(client, call, fetch_worker_config),
         "GetClusterInfo" => rpc_call!(client, call, get_cluster_info),
         "GetCurrentDeployment" => rpc_call!(client, call, get_current_deployment),
         "GetDeploymentReachability" => rpc_call!(client, call, get_deployment_reachability),
@@ -419,6 +416,8 @@ async fn call_workflow_service(
         }
         "UpdateNamespace" => rpc_call_on_trait!(client, call, WorkflowService, update_namespace),
         "UpdateSchedule" => rpc_call!(client, call, update_schedule),
+        "UpdateTaskQueueConfig" => rpc_call!(client, call, update_task_queue_config),
+        "UpdateWorkerConfig" => rpc_call!(client, call, update_worker_config),
         "UpdateWorkerDeploymentVersionMetadata" => {
             rpc_call!(client, call, update_worker_deployment_version_metadata)
         }

core/src/pollers/poll_buffer.rs

@@ -773,7 +773,6 @@ mod tests {
             });
         mock_client
             .expect_poll_workflow_task()
-            .times(1)
             .returning(move |_, _| async { Ok(Default::default()) }.boxed());
 
         let pb = LongPollBuffer::new_workflow_task(