Deal with bridge refactor merge conflicts

Author: Sushisource

packages/core-bridge/bridge-macros/src/lib.rs

@@ -9,6 +9,9 @@ use proc_macro::TokenStream;
 use syn::{DeriveInput, parse_macro_input};
 
 /// Procedural macro for defining bridge types with compile-time field name conversion
+///
+/// Note that enum types must all be defined on the JS side as objects with a `type` field which
+/// is the kebab-case representation of the enum variant.
 #[proc_macro_derive(TryFromJs)]
 pub fn try_from_js(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as DeriveInput);

packages/core-bridge/src/worker.rs

@@ -32,7 +32,7 @@ use crate::{
     runtime::{Runtime, RuntimeExt},
 };
 
-pub fn init(cx: &mut neon::prelude::ModuleContext) -> neon::prelude::NeonResult<()> {
+pub fn init(cx: &mut ModuleContext) -> NeonResult<()> {
     cx.export_function("newWorker", worker_new)?;
     cx.export_function("workerValidate", worker_validate)?;
 
@@ -412,19 +412,27 @@ mod config {
         api::worker::{
             ActivitySlotKind, LocalActivitySlotKind, PollerBehavior as CorePollerBehavior,
             SlotKind, WorkerConfig, WorkerConfigBuilder, WorkerConfigBuilderError,
-            WorkflowSlotKind,
+            WorkerDeploymentOptions as CoreWorkerDeploymentOptions,
+            WorkerDeploymentVersion as CoreWorkerDeploymentVersion, WorkflowSlotKind,
         },
+        protos::temporal::api::enums::v1::VersioningBehavior as CoreVersioningBehavior,
     };
 
-    use bridge_macros::TryFromJs;
-
     use super::custom_slot_supplier::CustomSlotSupplierOptions;
+    use crate::helpers::TryIntoJs;
+    use bridge_macros::TryFromJs;
+    use neon::context::Context;
+    use neon::object::Object;
+    use neon::prelude::JsResult;
+    use neon::types::JsObject;
+    use temporal_sdk_core::api::worker::WorkerVersioningStrategy;
 
     #[derive(TryFromJs)]
     pub struct BridgeWorkerOptions {
         identity: String,
         build_id: String,
         use_versioning: bool,
+        worker_deployment_options: Option<WorkerDeploymentOptions>,
         task_queue: String,
         namespace: String,
         tuner: WorkerTuner,
@@ -453,14 +461,44 @@ mod config {
         },
     }
 
+    #[derive(TryFromJs)]
+    pub struct WorkerDeploymentOptions {
+        version: WorkerDeploymentVersion,
+        use_worker_versioning: bool,
+        default_versioning_behavior: VersioningBehavior,
+    }
+
+    #[derive(TryFromJs)]
+    pub struct WorkerDeploymentVersion {
+        build_id: String,
+        deployment_name: String,
+    }
+
+    #[derive(TryFromJs)]
+    pub enum VersioningBehavior {
+        Pinned,
+        AutoUpgrade,
+    }
+
     impl BridgeWorkerOptions {
         pub(crate) fn into_core_config(self) -> Result<WorkerConfig, WorkerConfigBuilderError> {
             // Set all other options
             let mut builder = WorkerConfigBuilder::default();
             builder
                 .client_identity_override(Some(self.identity))
-                .worker_build_id(self.build_id)
-                .use_worker_versioning(self.use_versioning)
+                .versioning_strategy({
+                    if let Some(dopts) = self.worker_deployment_options {
+                        WorkerVersioningStrategy::WorkerDeploymentBased(dopts.into())
+                    } else if self.use_versioning {
+                        WorkerVersioningStrategy::LegacyBuildIdBased {
+                            build_id: self.build_id,
+                        }
+                    } else {
+                        WorkerVersioningStrategy::None {
+                            build_id: self.build_id,
+                        }
+                    }
+                })
                 .task_queue(self.task_queue)
                 .namespace(self.namespace)
                 .tuner(self.tuner.into_core_config()?)
@@ -483,13 +521,13 @@ mod config {
         fn from(val: PollerBehavior) -> Self {
             match val {
                 PollerBehavior::SimpleMaximum { maximum } => {
-                    CorePollerBehavior::SimpleMaximum(maximum)
+                    Self::SimpleMaximum(maximum)
                 }
                 PollerBehavior::Autoscaling {
                     minimum,
                     maximum,
                     initial,
-                } => CorePollerBehavior::Autoscaling {
+                } => Self::Autoscaling {
                     minimum,
                     maximum,
                     initial,
@@ -498,6 +536,56 @@ mod config {
         }
     }
 
+    impl From<WorkerDeploymentOptions> for CoreWorkerDeploymentOptions {
+        fn from(val: WorkerDeploymentOptions) -> Self {
+            Self {
+                version: val.version.into(),
+                use_worker_versioning: val.use_worker_versioning,
+                default_versioning_behavior: Some(val.default_versioning_behavior.into()),
+            }
+        }
+    }
+
+    impl From<WorkerDeploymentVersion> for CoreWorkerDeploymentVersion {
+        fn from(val: WorkerDeploymentVersion) -> Self {
+            Self {
+                build_id: val.build_id,
+                deployment_name: val.deployment_name,
+            }
+        }
+    }
+
+    impl From<CoreWorkerDeploymentVersion> for WorkerDeploymentVersion {
+        fn from(val: CoreWorkerDeploymentVersion) -> Self {
+            Self {
+                build_id: val.build_id,
+                deployment_name: val.deployment_name,
+            }
+        }
+    }
+
+    impl TryIntoJs for WorkerDeploymentVersion {
+        type Output = JsObject;
+
+        fn try_into_js<'cx>(self, cx: &mut impl Context<'cx>) -> JsResult<'cx, Self::Output> {
+            let obj = cx.empty_object();
+            let bid = self.build_id.try_into_js(cx)?;
+            obj.set(cx, "buildId", bid)?;
+            let dn = self.deployment_name.try_into_js(cx)?;
+            obj.set(cx, "deploymentName", dn)?;
+            Ok(obj)
+        }
+    }
+
+    impl From<VersioningBehavior> for CoreVersioningBehavior {
+        fn from(val: VersioningBehavior) -> Self {
+            match val {
+                VersioningBehavior::Pinned => Self::Pinned,
+                VersioningBehavior::AutoUpgrade => Self::AutoUpgrade,
+            }
+        }
+    }
+
     #[derive(TryFromJs)]
     #[allow(clippy::struct_field_names)]
     pub(super) struct WorkerTuner {
@@ -613,7 +701,7 @@ mod custom_slot_supplier {
     use tracing::warn;
 
     use crate::helpers::*;
-
+    use crate::worker::config::WorkerDeploymentVersion;
     // Custom Slot Supplier ////////////////////////////////////////////////////////////////////////////
 
     pub(super) struct SlotSupplierBridge<SK: SlotKind + Send + Sync + 'static> {
@@ -639,7 +727,10 @@ mod custom_slot_supplier {
                     slot_type: SK::kind().into(),
                     task_queue: ctx.task_queue().to_string(),
                     worker_identity: ctx.worker_identity().to_string(),
-                    worker_build_id: ctx.worker_build_id().to_string(),
+                    worker_deployment_version: ctx
+                        .worker_deployment_version()
+                        .clone()
+                        .map(Into::into),
                     is_sticky: ctx.is_sticky(),
                 };
 
@@ -680,7 +771,10 @@ mod custom_slot_supplier {
                 slot_type: SK::kind().into(),
                 task_queue: ctx.task_queue().to_string(),
                 worker_identity: ctx.worker_identity().to_string(),
-                worker_build_id: ctx.worker_build_id().to_string(),
+                worker_deployment_version: ctx
+                    .worker_deployment_version()
+                    .clone()
+                    .map(Into::into),
                 is_sticky: ctx.is_sticky(),
             };
 
@@ -809,7 +903,7 @@ mod custom_slot_supplier {
         slot_type: SlotKindType,
         task_queue: String,
         worker_identity: String,
-        worker_build_id: String,
+        worker_deployment_version: Option<WorkerDeploymentVersion>,
         is_sticky: bool,
     }
 

packages/core-bridge/ts/native.ts

@@ -41,6 +41,7 @@ export type JsonString<_T> = string;
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 
 export declare function newRuntime(telemOptions: RuntimeOptions): Runtime;
+
 export declare function runtimeShutdown(runtime: Runtime): void;
 
 export interface Runtime {
@@ -98,9 +99,13 @@ export interface OtelMetricsExporterOptions {
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 
 export declare function newClient(runtime: Runtime, clientOptions: ClientOptions): Promise<Client>;
+
 export declare function clientUpdateHeaders(client: Client, headers: Record<string, string>): void;
+
 export declare function clientUpdateApiKey(client: Client, apiKey: string): void;
+
 export declare function clientSendRequest(client: Client, call: RpcCall): Promise<Buffer>;
+
 export declare function clientClose(client: Client): void;
 
 export interface Client {
@@ -155,16 +160,21 @@ export interface RpcCall {
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 
 export declare function newWorker(client: Client, workerOptions: WorkerOptions): Worker;
+
 export declare function workerValidate(worker: Worker): Promise<void>;
 
 export declare function workerPollWorkflowActivation(worker: Worker): Promise<Buffer>;
+
 export declare function workerCompleteWorkflowActivation(worker: Worker, result: Buffer): Promise<void>;
 
 export declare function workerPollActivityTask(worker: Worker): Promise<Buffer>;
+
 export declare function workerCompleteActivityTask(worker: Worker, result: Buffer): Promise<void>;
+
 export declare function workerRecordActivityHeartbeat(worker: Worker, heartbeat: Buffer): void;
 
 export declare function workerInitiateShutdown(worker: Worker): void;
+
 export declare function workerFinalizeShutdown(worker: Worker): Promise<void>;
 
 export interface Worker {
@@ -175,6 +185,7 @@ export interface WorkerOptions {
   identity: string;
   buildId: string;
   useVersioning: boolean;
+  workerDeploymentOptions: Option<WorkerDeploymentOptions>;
   taskQueue: string;
   namespace: string;
   tuner: WorkerTunerOptions;
@@ -203,6 +214,19 @@ export type PollerBehavior =
       initial: number;
     };
 
+export type WorkerDeploymentOptions = {
+  version: WorkerDeploymentVersion;
+  useWorkerVersioning: boolean;
+  defaultVersioningBehavior: VersioningBehavior;
+};
+
+export type WorkerDeploymentVersion = {
+  buildId: string;
+  deploymentName: string;
+};
+
+export type VersioningBehavior = { type: 'pinned' } | { type: 'auto-upgrade' };
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 // Worker Tuner
 ////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -238,9 +262,13 @@ interface ResourceBasedTunerOptions {
 
 export interface CustomSlotSupplierOptions<SI extends SlotInfo> {
   type: 'custom';
+
   reserveSlot(ctx: SlotReserveContext, abortSignal: AbortSignal): Promise<SlotPermit>;
+
   tryReserveSlot(ctx: SlotReserveContext): Option<SlotPermit>;
+
   markSlotUsed(ctx: SlotMarkUsedContext<SI>): void;
+
   releaseSlot(ctx: SlotReleaseContext<SI>): void;
 }
 
@@ -268,7 +296,7 @@ export interface SlotReserveContext {
   slotType: SlotInfo['type'];
   taskQueue: string;
   workerIdentity: string;
-  workerBuildId: string;
+  workerDeploymentVersion: Option<WorkerDeploymentVersion>;
   isSticky: boolean;
 }
 
@@ -290,7 +318,9 @@ export interface SlotPermit {}
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 
 export declare function newReplayWorker(runtime: Runtime, workerOptions: WorkerOptions): [Worker, HistoryPusher];
+
 export declare function pushHistory(pusher: HistoryPusher, workflowId: string, history: Buffer): Promise<void>;
+
 export declare function closeHistoryStream(pusher: HistoryPusher): void;
 
 export interface HistoryPusher {
@@ -302,7 +332,9 @@ export interface HistoryPusher {
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 
 export declare function newEphemeralServer(runtime: Runtime, config: EphemeralServerConfig): Promise<EphemeralServer>;
+
 export declare function ephemeralServerGetTarget(server: EphemeralServer): string;
+
 export declare function ephemeralServerShutdown(server: EphemeralServer): Promise<void>;
 
 export interface EphemeralServer {

packages/test/src/test-bridge.ts

@@ -260,6 +260,7 @@ const GenericConfigs = {
       taskQueue: 'default',
       identity: 'test-worker',
       buildId: 'test-build-id',
+      workerDeploymentOptions: null,
       useVersioning: false,
       namespace: 'default',
       tuner: {

packages/testing/src/index.ts

@@ -14,6 +14,7 @@ export {
   TestWorkflowEnvironment,
   type LocalTestWorkflowEnvironmentOptions,
   type TimeSkippingTestWorkflowEnvironmentOptions,
+  type ExistingServerTestWorkflowEnvironmentOptions,
 } from './testing-workflow-environment';
 
 export {