Add Build ID to workflow info (#1333)

Co-authored-by: James Watkins-Harvey <james.watkinsharvey@temporal.io>
Co-authored-by: James Watkins-Harvey <mjameswh@users.noreply.github.com>

Author: Sushisource

packages/core-bridge/Cargo.lock

@@ -37,22 +37,22 @@ impl ArrayHandleConversionsExt for Handle<'_, JsArray> {
         let len = js_vec.len();
         let mut ret_vec = Vec::<String>::with_capacity(len);
 
-        for i in 0..len {
-            ret_vec.push(js_vec[i].downcast_or_throw::<JsString, _>(cx)?.value(cx));
+        for i in js_vec.iter().take(len) {
+            ret_vec.push(i.downcast_or_throw::<JsString, _>(cx)?.value(cx));
         }
         Ok(ret_vec)
     }
 }
 
-type TelemOptsRes = NeonResult<(
+pub(crate) type TelemOptsRes = (
     TelemetryOptions,
     Option<Box<dyn FnOnce() -> Arc<dyn CoreMeter> + Send>>,
-)>;
+);
 
 pub trait ObjectHandleConversionsExt {
     fn set_default(&self, cx: &mut FunctionContext, key: &str, value: &str) -> NeonResult<()>;
     fn as_client_options(&self, ctx: &mut FunctionContext) -> NeonResult<ClientOptions>;
-    fn as_telemetry_options(&self, cx: &mut FunctionContext) -> TelemOptsRes;
+    fn as_telemetry_options(&self, cx: &mut FunctionContext) -> NeonResult<TelemOptsRes>;
     fn as_worker_config(&self, cx: &mut FunctionContext) -> NeonResult<WorkerConfig>;
     fn as_ephemeral_server_config(
         &self,
@@ -169,7 +169,7 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
             .expect("Core server gateway options must be valid"))
     }
 
-    fn as_telemetry_options(&self, cx: &mut FunctionContext) -> TelemOptsRes {
+    fn as_telemetry_options(&self, cx: &mut FunctionContext) -> NeonResult<TelemOptsRes> {
         let mut telemetry_opts = TelemetryOptionsBuilder::default();
         if js_optional_value_getter!(cx, self, "noTemporalPrefixForMetrics", JsBoolean)
             .unwrap_or_default()

packages/core-bridge/sdk-core

@@ -11,8 +11,7 @@ use std::{
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 use temporal_client::{ClientInitError, ConfiguredClient, TemporalServiceClientWithMetrics};
-use temporal_sdk_core::api::telemetry::metrics::CoreMeter;
-use temporal_sdk_core::api::telemetry::{CoreTelemetry, TelemetryOptions};
+use temporal_sdk_core::api::telemetry::CoreTelemetry;
 use temporal_sdk_core::CoreRuntime;
 use temporal_sdk_core::{
     ephemeral_server::EphemeralServer as CoreEphemeralServer,
@@ -117,10 +116,7 @@ pub enum RuntimeRequest {
 /// Bridges requests from JS to core and sends responses back to JS using a neon::Channel.
 /// Blocks current thread until a [Shutdown] request is received in channel.
 pub fn start_bridge_loop(
-    telemetry_options: (
-        TelemetryOptions,
-        Option<Box<dyn FnOnce() -> Arc<dyn CoreMeter> + Send>>,
-    ),
+    telemetry_options: TelemOptsRes,
     channel: Arc<Channel>,
     receiver: &mut UnboundedReceiver<RuntimeRequest>,
 ) {

packages/core-bridge/src/conversions.rs

@@ -708,6 +708,7 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
       historySize: result.historySize,
       startTime: result.startTime,
       runStartTime: result.runStartTime,
+      currentBuildId: result.currentBuildId,
       // unsafe.now is a function, so doesn't make it through serialization, but .now is required, so we need to cast
       unsafe: { isReplaying: false } as UnsafeWorkflowInfo,
     });

packages/core-bridge/src/runtime.rs

@@ -5,6 +5,7 @@ import * as activity from '@temporalio/activity';
 import { tsToMs } from '@temporalio/common/lib/time';
 import { CancelReason } from '@temporalio/worker/lib/activity';
 import * as workflow from '@temporalio/workflow';
+import { defineQuery, defineSignal } from '@temporalio/workflow';
 import { signalSchedulingWorkflow } from './activities/helpers';
 import { activityStartedSignal } from './workflows/definitions';
 import * as workflows from './workflows';
@@ -226,7 +227,10 @@ test('Start of workflow with signal is delayed', async (t) => {
 export async function executeEagerActivity(): Promise<void> {
   const scheduleActivity = () =>
     workflow
-      .proxyActivities({ scheduleToCloseTimeout: '5s', allowEagerDispatch: true })
+      .proxyActivities({
+        scheduleToCloseTimeout: '5s',
+        allowEagerDispatch: true,
+      })
       .testActivity()
       .then((res) => {
         if (res !== 'workflow-and-activity-worker')
@@ -305,3 +309,53 @@ test("Worker doesn't request Eager Activity Dispatch if no activities are regist
   t.is(activityTaskStarted?.length, 1);
   t.is(activityTaskStarted?.[0]?.activityTaskStartedEventAttributes?.attempt, 1);
 });
+
+const unblockSignal = defineSignal('unblock');
+const getBuildIdQuery = defineQuery<string>('getBuildId');
+
+export async function buildIdTester(): Promise<void> {
+  let blocked = true;
+  workflow.setHandler(unblockSignal, () => {
+    blocked = false;
+  });
+
+  workflow.setHandler(getBuildIdQuery, () => {
+    return workflow.workflowInfo().currentBuildId ?? '';
+  });
+
+  // The unblock signal will only be sent once we are in Worker 1.1.
+  // Therefore, up to this point, we are runing in Worker 1.0
+  await workflow.condition(() => !blocked);
+  // From this point on, we are runing in Worker 1.1
+
+  // Prevent workflow completion
+  await workflow.condition(() => false);
+}
+
+test('Build Id appropriately set in workflow info', async (t) => {
+  const { taskQueue, createWorker } = helpers(t);
+  const wfid = `${taskQueue}-` + randomUUID();
+  const client = t.context.env.client;
+
+  const worker1 = await createWorker({ buildId: '1.0' });
+  await worker1.runUntil(async () => {
+    const handle = await client.workflow.start(buildIdTester, {
+      taskQueue,
+      workflowId: wfid,
+    });
+    t.is(await handle.query(getBuildIdQuery), '1.0');
+  });
+
+  await client.workflowService.resetStickyTaskQueue({
+    namespace: worker1.options.namespace,
+    execution: { workflowId: wfid },
+  });
+
+  const worker2 = await createWorker({ buildId: '1.1' });
+  await worker2.runUntil(async () => {
+    const handle = await client.workflow.getHandle(wfid);
+    t.is(await handle.query(getBuildIdQuery), '1.0');
+    await handle.signal(unblockSignal);
+    t.is(await handle.query(getBuildIdQuery), '1.1');
+  });
+});

packages/test/src/integration-tests-old.ts

@@ -116,7 +116,7 @@ if (RUN_INTEGRATION_TESTS) {
     });
 
     // Capture volatile values that are hard to predict
-    const { historySize, startTime, runStartTime } = recordedCalls[0].info;
+    const { historySize, startTime, runStartTime, currentBuildId } = recordedCalls[0].info;
     t.true(historySize > 300);
 
     const info: WorkflowInfo = {
@@ -146,6 +146,7 @@ if (RUN_INTEGRATION_TESTS) {
       historySize,
       startTime,
       runStartTime,
+      currentBuildId,
       // unsafe.now() doesn't make it through serialization, but .now is required, so we need to cast
       unsafe: {
         isReplaying: false,

packages/test/src/test-integration-workflows.ts

@@ -1224,6 +1224,7 @@ export class Worker {
       // A zero value means that it was not set by the server
       historySize: activation.historySizeBytes.toNumber(),
       continueAsNewSuggested: activation.continueAsNewSuggested,
+      currentBuildId: activation.buildIdForCurrentTask,
       unsafe: {
         now: () => Date.now(), // re-set in initRuntime
         isReplaying: activation.isReplaying,

packages/test/src/test-sinks.ts

@@ -162,6 +162,14 @@ export interface WorkflowInfo {
    */
   readonly cronScheduleToScheduleInterval?: number;
 
+  /**
+   * The Build ID of the worker which executed the current Workflow Task. May be undefined if the
+   * task was completed by a worker without a Build ID. If this worker is the one executing this
+   * task for the first time and has a Build ID set, then its ID will be used. This value may change
+   * over the lifetime of the workflow run, but is deterministic and safe to use for branching.
+   */
+  readonly currentBuildId?: string;
+
   readonly unsafe: UnsafeWorkflowInfo;
 }
 

packages/worker/src/worker.ts

@@ -199,6 +199,7 @@ export function activate(activation: coresdk.workflow_activation.WorkflowActivat
         // historySize === 0 means WFT was generated by pre-1.20.0 server, and the history size is unknown
         historySize: activation.historySizeBytes?.toNumber() || 0,
         continueAsNewSuggested: activation.continueAsNewSuggested ?? false,
+        currentBuildId: activation.buildIdForCurrentTask ?? undefined,
         unsafe: {
           ...info.unsafe,
           isReplaying: activation.isReplaying ?? false,