Add Fairness keys & weights

Author: Sushisource

packages/common/src/priority.ts

@@ -11,7 +11,8 @@ import type { temporal } from '@temporalio/proto';
  * 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)
+ * 2. Then, consider fairness: the fairness mechanism attempts to dispatch tasks for a given key in
+ *    proportion to its weight.
  */
 export interface Priority {
   /**
@@ -26,13 +27,43 @@ export interface Priority {
    * The default priority is (min+max)/2. With the default max of 5 and min of 1, that comes out to 3.
    */
   priorityKey?: number;
+
+  /**
+   * FairnessKey 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.
+   *
+   * Fairness keys are limited to 64 bytes.
+   */
+  fairnessKey?: string;
+
+  /**
+   * FairnessWeight 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 to the range [0.001, 1000].
+   */
+  fairnessWeight?: number;
 }
 
 /**
  * Turn a proto compatible Priority into a TS Priority
  */
 export function decodePriority(priority?: temporal.api.common.v1.IPriority | null): Priority {
-  return { priorityKey: priority?.priorityKey ?? undefined };
+  return {
+    priorityKey: priority?.priorityKey ?? undefined,
+    fairnessKey: priority?.fairnessKey ?? undefined,
+    fairnessWeight: priority?.fairnessWeight ?? undefined,
+  };
 }
 
 /**
@@ -50,5 +81,7 @@ export function compilePriority(priority: Priority): temporal.api.common.v1.IPri
 
   return {
     priorityKey: priority.priorityKey ?? 0,
+    fairnessKey: priority.fairnessKey ?? '',
+    fairnessWeight: priority.fairnessWeight ?? 0,
   };
 }

packages/test/src/helpers.ts

@@ -308,3 +308,14 @@ export async function saveHistory(fname: string, history: iface.temporal.api.his
   const fpath = path.resolve(__dirname, `../history_files/${fname}`);
   await fs.writeFile(fpath, historyToJSON(history));
 }
+
+export function approximatelyEqual(
+  a: number | null | undefined,
+  b: number | null | undefined,
+  tolerance = Number.EPSILON
+): boolean {
+  if (a === null || a === undefined || b === null || b === undefined) {
+    return false;
+  }
+  return Math.abs(a - b) < tolerance;
+}

packages/test/src/test-integration-split-three.ts

@@ -8,7 +8,7 @@ import { configMacro, makeTestFn } from './helpers-integration-multi-codec';
 import { configurableHelpers } from './helpers-integration';
 import { withZeroesHTTPServer } from './zeroes-http-server';
 import * as activities from './activities';
-import { cleanOptionalStackTrace } from './helpers';
+import { approximatelyEqual, cleanOptionalStackTrace } from './helpers';
 import * as workflows from './workflows';
 
 const test = makeTestFn(() => bundleWorkflowCode({ workflowsPath: require.resolve('./workflows') }));
@@ -151,7 +151,7 @@ test(
     const worker = await createWorkerWithDefaults(t, { activities });
     const handle = await startWorkflow(workflows.priorityWorkflow, {
       args: [false, 1],
-      priority: { priorityKey: 1 },
+      priority: { priorityKey: 1, fairnessKey: 'main-workflow', fairnessWeight: 3.0 },
     });
     await worker.runUntil(handle.result());
     let firstChild = true;
@@ -161,19 +161,28 @@ test(
       switch (event.eventType) {
         case temporal.api.enums.v1.EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED:
           t.deepEqual(event.workflowExecutionStartedEventAttributes?.priority?.priorityKey, 1);
+          t.deepEqual(event.workflowExecutionStartedEventAttributes?.priority?.fairnessKey, 'main-workflow');
+          t.deepEqual(event.workflowExecutionStartedEventAttributes?.priority?.fairnessWeight, 3.0);
           break;
         case temporal.api.enums.v1.EventType.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED: {
-          const pri = event.startChildWorkflowExecutionInitiatedEventAttributes?.priority?.priorityKey;
+          const priority = event.startChildWorkflowExecutionInitiatedEventAttributes?.priority;
           if (firstChild) {
-            t.deepEqual(pri, 4);
+            t.deepEqual(priority?.priorityKey, 4);
+            t.deepEqual(priority?.fairnessKey, 'child-workflow-1');
+            t.deepEqual(priority?.fairnessWeight, 2.5);
             firstChild = false;
           } else {
-            t.deepEqual(pri, 2);
+            t.deepEqual(priority?.priorityKey, 2);
+            t.deepEqual(priority?.fairnessKey, 'child-workflow-2');
+            t.deepEqual(priority?.fairnessWeight, 1.0);
           }
           break;
         }
         case temporal.api.enums.v1.EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED:
           t.deepEqual(event.activityTaskScheduledEventAttributes?.priority?.priorityKey, 5);
+          t.deepEqual(event.activityTaskScheduledEventAttributes?.priority?.fairnessKey, 'fair-activity');
+          // For some insane reason when proto reads this event it mangles the number to 4.19999999 something. Thanks Javascript.
+          t.assert(approximatelyEqual(event.activityTaskScheduledEventAttributes?.priority?.fairnessWeight, 4.2));
           break;
       }
     }

packages/test/src/workflows/priority.ts

@@ -1,7 +1,10 @@
 import { executeChild, proxyActivities, startChild, workflowInfo } from '@temporalio/workflow';
 import type * as activities from '../activities';
 
-const { echo } = proxyActivities<typeof activities>({ startToCloseTimeout: '5s', priority: { priorityKey: 5 } });
+const { echo } = proxyActivities<typeof activities>({
+  startToCloseTimeout: '5s',
+  priority: { priorityKey: 5, fairnessKey: 'fair-activity', fairnessWeight: 4.2 },
+});
 
 export async function priorityWorkflow(stopAfterCheck: boolean, expectedPriority: number | undefined): Promise<void> {
   const info = workflowInfo();
@@ -17,9 +20,15 @@ export async function priorityWorkflow(stopAfterCheck: boolean, expectedPriority
     return;
   }
 
-  await executeChild(priorityWorkflow, { args: [true, 4], priority: { priorityKey: 4 } });
+  await executeChild(priorityWorkflow, {
+    args: [true, 4],
+    priority: { priorityKey: 4, fairnessKey: 'child-workflow-1', fairnessWeight: 2.5 },
+  });
 
-  const child = await startChild(priorityWorkflow, { args: [true, 2], priority: { priorityKey: 2 } });
+  const child = await startChild(priorityWorkflow, {
+    args: [true, 2],
+    priority: { priorityKey: 2, fairnessKey: 'child-workflow-2', fairnessWeight: 1.0 },
+  });
   await child.result();
 
   await echo('hi');