SDK Worker Executes activities

crates/client/src/lib.rs

@@ -170,6 +170,8 @@ pub struct ConnectionOptions {
     #[builder(default = "temporal-rust".to_owned())]
     #[cfg_attr(feature = "core-based-sdk", builder(setters(vis = "pub")))]
     client_name: String,
+    // TODO [rust-sdk-branch]: SDK should set this to its version. Doing that probably easiest
+    // after adding proper client interceptors.
     /// The version of the SDK being implemented on top of core. Is set as `client-version` header
     /// in all RPC calls. The server decides if the client is supported based on this.
     #[builder(default = VERSION.to_owned())]

crates/common/Cargo.toml

@@ -26,6 +26,7 @@ anyhow = "1.0"
 async-trait = "0.1"
 base64 = "0.22"
 bon = { workspace = true }
+crc32fast = "1"
 dirs = { version = "6.0", optional = true }
 derive_more = { workspace = true }
 erased-serde = "0.4"

crates/common/src/activity_definition.rs

@@ -9,7 +9,7 @@ use crate::data_converters::{TemporalDeserializable, TemporalSerializable};
 /// Implement on a marker struct to define an activity.
 pub trait ActivityDefinition {
     /// Type of the input argument to the workflow
-    type Input: TemporalDeserializable + 'static;
+    type Input: TemporalDeserializable + TemporalSerializable + 'static;
     /// Type of the output of the workflow
     type Output: TemporalSerializable + 'static;
 

crates/common/src/data_converters.rs

@@ -47,9 +47,28 @@ impl PayloadConverter {
     // TODO [rust-sdk-branch]: Proto binary, other standard built-ins
 }
 
+#[derive(Debug)]
 pub enum PayloadConversionError {
     WrongEncoding,
-    EncodingError(Box<dyn std::error::Error>),
+    EncodingError(Box<dyn std::error::Error + Send + Sync>),
+}
+
+impl std::fmt::Display for PayloadConversionError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            PayloadConversionError::WrongEncoding => write!(f, "Wrong encoding"),
+            PayloadConversionError::EncodingError(err) => write!(f, "Encoding error: {}", err),
+        }
+    }
+}
+
+impl std::error::Error for PayloadConversionError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            PayloadConversionError::WrongEncoding => None,
+            PayloadConversionError::EncodingError(err) => Some(err.as_ref()),
+        }
+    }
 }
 
 pub trait FailureConverter {
@@ -71,13 +90,13 @@ pub struct DefaultFailureConverter;
 pub trait PayloadCodec {
     fn encode(
         &self,
-        payloads: Vec<Payload>,
         context: &SerializationContext,
+        payloads: Vec<Payload>,
     ) -> BoxFuture<'static, Vec<Payload>>;
     fn decode(
         &self,
-        payloads: Vec<Payload>,
         context: &SerializationContext,
+        payloads: Vec<Payload>,
     ) -> BoxFuture<'static, Vec<Payload>>;
 }
 pub struct DefaultPayloadCodec;
@@ -94,16 +113,16 @@ pub trait TemporalSerializable {
         None
     }
 }
-///
+
 /// Indicates some type can be deserialized for use with Temporal.
 ///
 /// You don't need to implement this unless you are using a non-serde-compatible custom converter,
 /// in which case you should implement the to/from_payload functions on some wrapper type.
 pub trait TemporalDeserializable: Sized {
     fn from_serde(
         _: &dyn ErasedSerdePayloadConverter,
-        _: Payload,
         _: &SerializationContext,
+        _: Payload,
     ) -> Option<Self> {
         None
     }
@@ -122,26 +141,26 @@ pub struct RawValue {
 pub trait GenericPayloadConverter {
     fn to_payload<T: TemporalSerializable + 'static>(
         &self,
-        val: &T,
         context: &SerializationContext,
+        val: &T,
     ) -> Result<Payload, PayloadConversionError>;
     #[allow(clippy::wrong_self_convention)]
     fn from_payload<T: TemporalDeserializable + 'static>(
         &self,
-        payload: Payload,
         context: &SerializationContext,
+        payload: Payload,
     ) -> Result<T, PayloadConversionError>;
 }
 
 impl GenericPayloadConverter for PayloadConverter {
     fn to_payload<T: TemporalSerializable + 'static>(
         &self,
-        val: &T,
         context: &SerializationContext,
+        val: &T,
     ) -> Result<Payload, PayloadConversionError> {
         match self {
             PayloadConverter::Serde(pc) => {
-                Ok(pc.to_payload(val.as_serde().ok_or_else(|| todo!())?, context)?)
+                Ok(pc.to_payload(context, val.as_serde().ok_or_else(|| todo!())?)?)
             }
             PayloadConverter::UseWrappers => {
                 Ok(T::to_payload(val, context).ok_or_else(|| todo!())?)
@@ -153,12 +172,12 @@ impl GenericPayloadConverter for PayloadConverter {
 
     fn from_payload<T: TemporalDeserializable + 'static>(
         &self,
-        payload: Payload,
         context: &SerializationContext,
+        payload: Payload,
     ) -> Result<T, PayloadConversionError> {
         match self {
             PayloadConverter::Serde(pc) => {
-                Ok(T::from_serde(pc.as_ref(), payload, context).ok_or_else(|| todo!())?)
+                Ok(T::from_serde(pc.as_ref(), context, payload).ok_or_else(|| todo!())?)
             }
             PayloadConverter::UseWrappers => {
                 Ok(T::from_payload(payload, context).ok_or_else(|| todo!())?)
@@ -183,22 +202,22 @@ where
 {
     fn from_serde(
         pc: &dyn ErasedSerdePayloadConverter,
-        payload: Payload,
         context: &SerializationContext,
+        payload: Payload,
     ) -> Option<Self>
     where
         Self: Sized,
     {
-        erased_serde::deserialize(&mut pc.from_payload(payload, context).ok()?).ok()
+        erased_serde::deserialize(&mut pc.from_payload(context, payload).ok()?).ok()
     }
 }
 
 struct SerdeJsonPayloadConverter;
 impl ErasedSerdePayloadConverter for SerdeJsonPayloadConverter {
     fn to_payload(
         &self,
+        _: &SerializationContext,
         value: &dyn erased_serde::Serialize,
-        _context: &SerializationContext,
     ) -> Result<Payload, PayloadConversionError> {
         let as_json = serde_json::to_vec(value).map_err(|_| todo!())?;
         Ok(Payload {
@@ -213,8 +232,8 @@ impl ErasedSerdePayloadConverter for SerdeJsonPayloadConverter {
 
     fn from_payload(
         &self,
+        _: &SerializationContext,
         payload: Payload,
-        _context: &SerializationContext,
     ) -> Result<Box<dyn erased_serde::Deserializer<'static>>, PayloadConversionError> {
         // TODO: Would check metadata
         let json_v: serde_json::Value =
@@ -225,14 +244,14 @@ impl ErasedSerdePayloadConverter for SerdeJsonPayloadConverter {
 pub trait ErasedSerdePayloadConverter: Send + Sync {
     fn to_payload(
         &self,
-        value: &dyn erased_serde::Serialize,
         context: &SerializationContext,
+        value: &dyn erased_serde::Serialize,
     ) -> Result<Payload, PayloadConversionError>;
     #[allow(clippy::wrong_self_convention)]
     fn from_payload(
         &self,
-        payload: Payload,
         context: &SerializationContext,
+        payload: Payload,
     ) -> Result<Box<dyn erased_serde::Deserializer<'static>>, PayloadConversionError>;
 }
 
@@ -299,15 +318,15 @@ impl FailureConverter for DefaultFailureConverter {
 impl PayloadCodec for DefaultPayloadCodec {
     fn encode(
         &self,
-        payloads: Vec<Payload>,
         _: &SerializationContext,
+        payloads: Vec<Payload>,
     ) -> BoxFuture<'static, Vec<Payload>> {
         async move { payloads }.boxed()
     }
     fn decode(
         &self,
-        payloads: Vec<Payload>,
         _: &SerializationContext,
+        payloads: Vec<Payload>,
     ) -> BoxFuture<'static, Vec<Payload>> {
         async move { payloads }.boxed()
     }

crates/common/src/worker.rs

@@ -2,7 +2,12 @@
 //! with workers.
 
 use crate::protos::{coresdk, temporal, temporal::api::enums::v1::VersioningBehavior};
-use std::str::FromStr;
+use std::{
+    fs::File,
+    io::{self, BufReader, Read},
+    str::FromStr,
+    sync::OnceLock,
+};
 
 /// Specifies which task types a worker will poll for.
 ///
@@ -84,6 +89,19 @@ pub struct WorkerDeploymentOptions {
     pub default_versioning_behavior: Option<VersioningBehavior>,
 }
 
+impl WorkerDeploymentOptions {
+    pub fn from_build_id(build_id: String) -> Self {
+        Self {
+            version: WorkerDeploymentVersion {
+                deployment_name: "".to_owned(),
+                build_id,
+            },
+            use_worker_versioning: false,
+            default_versioning_behavior: None,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Eq, PartialEq, Hash)]
 pub struct WorkerDeploymentVersion {
     /// Name of the deployment
@@ -138,3 +156,33 @@ impl From<temporal::api::deployment::v1::WorkerDeploymentVersion> for WorkerDepl
         }
     }
 }
+
+static CACHED_BUILD_ID: OnceLock<String> = OnceLock::new();
+
+/// Build ID derived from hashing the on-disk bytes of the current executable.
+/// Deterministic across machines for the same binary. Cached per-process.
+pub fn build_id_from_current_exe() -> &'static str {
+    CACHED_BUILD_ID
+        .get_or_init(|| compute_crc32_exe_id().unwrap_or_else(|_| "undetermined".to_owned()))
+}
+
+fn compute_crc32_exe_id() -> io::Result<String> {
+    let exe_path = std::env::current_exe()?;
+    let file = File::open(exe_path)?;
+    let mut reader = BufReader::new(file);
+
+    let mut hasher = crc32fast::Hasher::new();
+    let mut buf = [0u8; 128 * 1024];
+
+    loop {
+        let n = reader.read(&mut buf)?;
+        if n == 0 {
+            break;
+        }
+        hasher.update(&buf[..n]);
+    }
+
+    let crc = hasher.finalize();
+
+    Ok(format!("{:08x}", crc))
+}