Eliminate worker trait & move worker config into core

Author: Sushisource

crates/common/src/errors.rs

@@ -12,228 +12,14 @@ mod activity_definition;
 pub mod data_converters;
 #[cfg(feature = "envconfig")]
 pub mod envconfig;
-pub mod errors;
 #[doc(hidden)]
 pub mod fsm_trait;
 pub mod protos;
 pub mod telemetry;
 pub mod worker;
 
-use crate::{
-    errors::{
-        CompleteActivityError, CompleteNexusError, CompleteWfError, PollError,
-        WorkerValidationError,
-    },
-    protos::coresdk::{
-        ActivityHeartbeat, ActivityTaskCompletion,
-        activity_task::ActivityTask,
-        nexus::{NexusTask, NexusTaskCompletion},
-        workflow_activation::WorkflowActivation,
-        workflow_completion::WorkflowActivationCompletion,
-    },
-    worker::WorkerConfig,
-};
-use std::sync::Arc;
-use uuid::Uuid;
-
 pub use activity_definition::ActivityDefinition;
 
-// TODO [rust-sdk-branch]: Move to core
-/// This trait is the primary way by which language specific SDKs interact with the core SDK.
-/// It represents one worker, which has a (potentially shared) client for connecting to the service
-/// and is bound to a specific task queue.
-#[async_trait::async_trait]
-pub trait Worker: Send + Sync {
-    /// Validate that the worker can properly connect to server, plus any other validation that
-    /// needs to be done asynchronously. Lang SDKs should call this function once before calling
-    /// any others.
-    async fn validate(&self) -> Result<(), WorkerValidationError>;
-
-    /// Ask the worker for some work, returning a [WorkflowActivation]. It is then the language
-    /// SDK's responsibility to call the appropriate workflow code with the provided inputs. Blocks
-    /// indefinitely until such work is available or [Worker::shutdown] is called.
-    ///
-    /// It is important to understand that all activations must be responded to. There can only
-    /// be one outstanding activation for a particular run of a workflow at any time. If an
-    /// activation is not responded to, it will cause that workflow to become stuck forever.
-    ///
-    /// See [WorkflowActivation] for more details on the expected behavior of lang w.r.t activation
-    /// & job processing.
-    ///
-    /// Do not call poll concurrently. It handles polling the server concurrently internally.
-    async fn poll_workflow_activation(&self) -> Result<WorkflowActivation, PollError>;
-
-    /// Ask the worker for some work, returning an [ActivityTask]. It is then the language SDK's
-    /// responsibility to call the appropriate activity code with the provided inputs. Blocks
-    /// indefinitely until such work is available or [Worker::shutdown] is called.
-    ///
-    /// Do not call poll concurrently. It handles polling the server concurrently internally.
-    async fn poll_activity_task(&self) -> Result<ActivityTask, PollError>;
-
-    /// Ask the worker for some nexus related work. It is then the language SDK's
-    /// responsibility to call the appropriate nexus operation handler code with the provided
-    /// inputs. Blocks indefinitely until such work is available or [Worker::shutdown] is called.
-    ///
-    /// All tasks must be responded to for shutdown to complete.
-    ///
-    /// Do not call poll concurrently. It handles polling the server concurrently internally.
-    async fn poll_nexus_task(&self) -> Result<NexusTask, PollError>;
-
-    /// Tell the worker that a workflow activation has completed. May (and should) be freely called
-    /// concurrently. The future may take some time to resolve, as fetching more events might be
-    /// necessary for completion to... complete - thus SDK implementers should make sure they do
-    /// not serialize completions.
-    async fn complete_workflow_activation(
-        &self,
-        completion: WorkflowActivationCompletion,
-    ) -> Result<(), CompleteWfError>;
-
-    /// Tell the worker that an activity has finished executing. May (and should) be freely called
-    /// concurrently.
-    async fn complete_activity_task(
-        &self,
-        completion: ActivityTaskCompletion,
-    ) -> Result<(), CompleteActivityError>;
-
-    /// Tell the worker that a nexus task has completed. May (and should) be freely called
-    /// concurrently.
-    async fn complete_nexus_task(
-        &self,
-        completion: NexusTaskCompletion,
-    ) -> Result<(), CompleteNexusError>;
-
-    /// Notify the Temporal service that an activity is still alive. Long running activities that
-    /// take longer than `activity_heartbeat_timeout` to finish must call this function in order to
-    /// report progress, otherwise the activity will timeout and a new attempt will be scheduled.
-    ///
-    /// The first heartbeat request will be sent immediately, subsequent rapid calls to this
-    /// function will result in heartbeat requests being aggregated and the last one received during
-    /// the aggregation period will be sent to the server, where that period is defined as half the
-    /// heartbeat timeout.
-    ///
-    /// Unlike Java/Go SDKs we do not return cancellation status as part of heartbeat response and
-    /// instead send it as a separate activity task to the lang, decoupling heartbeat and
-    /// cancellation processing.
-    ///
-    /// For now activity still need to send heartbeats if they want to receive cancellation
-    /// requests. In the future we will change this and will dispatch cancellations more
-    /// proactively. Note that this function does not block on the server call and returns
-    /// immediately. Underlying validation errors are swallowed and logged, this has been agreed to
-    /// be optimal behavior for the user as we don't want to break activity execution due to badly
-    /// configured heartbeat options.
-    fn record_activity_heartbeat(&self, details: ActivityHeartbeat);
-
-    /// Request that a workflow be evicted by its run id. This will generate a workflow activation
-    /// with the eviction job inside it to be eventually returned by
-    /// [Worker::poll_workflow_activation]. If the workflow had any existing outstanding activations,
-    /// such activations are invalidated and subsequent completions of them will do nothing and log
-    /// a warning.
-    fn request_workflow_eviction(&self, run_id: &str);
-
-    /// Return this worker's config
-    fn get_config(&self) -> &WorkerConfig;
-
-    /// Initiate shutdown. See [Worker::shutdown], this is just a sync version that starts the
-    /// process. You can then wait on `shutdown` or [Worker::finalize_shutdown].
-    fn initiate_shutdown(&self);
-
-    /// Initiates async shutdown procedure, eventually ceases all polling of the server and shuts
-    /// down this worker. [Worker::poll_workflow_activation] and [Worker::poll_activity_task] should
-    /// be called until both return a `ShutDown` error to ensure that all outstanding work is
-    /// complete. This means that the lang sdk will need to call
-    /// [Worker::complete_workflow_activation] and [Worker::complete_activity_task] for those
-    /// workflows & activities until they are done. At that point, the lang SDK can end the process,
-    /// or drop the [Worker] instance via [Worker::finalize_shutdown], which will close the
-    /// connection and free resources. If you have set [WorkerConfig::task_types] to exclude
-    /// [worker::WorkerTaskTypes::activity_only()], you may skip calling [Worker::poll_activity_task].
-    ///
-    /// Lang implementations should use [Worker::initiate_shutdown] followed by
-    /// [Worker::finalize_shutdown].
-    async fn shutdown(&self);
-
-    /// Completes shutdown and frees all resources. You should avoid simply dropping workers, as
-    /// this does not allow async tasks to report any panics that may have occurred cleanly.
-    ///
-    /// This should be called only after [Worker::shutdown] has resolved and/or both polling
-    /// functions have returned `ShutDown` errors.
-    async fn finalize_shutdown(self);
-
-    /// Unique identifier for this worker instance.
-    /// This must be stable across the worker's lifetime and unique per instance.
-    fn worker_instance_key(&self) -> Uuid;
-}
-
-#[async_trait::async_trait]
-impl<W> Worker for Arc<W>
-where
-    W: Worker + ?Sized,
-{
-    async fn validate(&self) -> Result<(), WorkerValidationError> {
-        (**self).validate().await
-    }
-
-    async fn poll_workflow_activation(&self) -> Result<WorkflowActivation, PollError> {
-        (**self).poll_workflow_activation().await
-    }
-
-    async fn poll_activity_task(&self) -> Result<ActivityTask, PollError> {
-        (**self).poll_activity_task().await
-    }
-
-    async fn poll_nexus_task(&self) -> Result<NexusTask, PollError> {
-        (**self).poll_nexus_task().await
-    }
-
-    async fn complete_workflow_activation(
-        &self,
-        completion: WorkflowActivationCompletion,
-    ) -> Result<(), CompleteWfError> {
-        (**self).complete_workflow_activation(completion).await
-    }
-
-    async fn complete_activity_task(
-        &self,
-        completion: ActivityTaskCompletion,
-    ) -> Result<(), CompleteActivityError> {
-        (**self).complete_activity_task(completion).await
-    }
-
-    async fn complete_nexus_task(
-        &self,
-        completion: NexusTaskCompletion,
-    ) -> Result<(), CompleteNexusError> {
-        (**self).complete_nexus_task(completion).await
-    }
-
-    fn record_activity_heartbeat(&self, details: ActivityHeartbeat) {
-        (**self).record_activity_heartbeat(details)
-    }
-
-    fn request_workflow_eviction(&self, run_id: &str) {
-        (**self).request_workflow_eviction(run_id)
-    }
-
-    fn get_config(&self) -> &WorkerConfig {
-        (**self).get_config()
-    }
-
-    fn initiate_shutdown(&self) {
-        (**self).initiate_shutdown()
-    }
-
-    async fn shutdown(&self) {
-        (**self).shutdown().await
-    }
-
-    async fn finalize_shutdown(self) {
-        panic!("Can't finalize shutdown on Arc'd worker")
-    }
-
-    fn worker_instance_key(&self) -> Uuid {
-        (**self).worker_instance_key()
-    }
-}
-
 macro_rules! dbg_panic {
   ($($arg:tt)*) => {
       use tracing::error;

crates/common/src/lib.rs

@@ -70,7 +70,7 @@ pub mod coresdk {
 
     #[allow(clippy::module_inception)]
     pub mod activity_task {
-        use crate::protos::{coresdk::ActivityTaskCompletion, task_token::fmt_tt};
+        use crate::protos::{coresdk::ActivityTaskCompletion, task_token::format_task_token};
         use std::fmt::{Display, Formatter};
         tonic::include_proto!("coresdk.activity_task");
 
@@ -119,7 +119,7 @@ pub mod coresdk {
                 write!(
                     f,
                     "ActivityTaskCompletion(token: {}",
-                    fmt_tt(&self.task_token),
+                    format_task_token(&self.task_token),
                 )?;
                 if let Some(r) = self.result.as_ref().and_then(|r| r.status.as_ref()) {
                     write!(f, ", {r}")?;

crates/common/src/protos/mod.rs

@@ -36,13 +36,13 @@ impl TaskToken {
 
 impl Display for TaskToken {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.write_str(&fmt_tt(&self.0))
+        f.write_str(&format_task_token(&self.0))
     }
 }
 
 impl Debug for TaskToken {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.write_str(&format!("TaskToken({})", fmt_tt(&self.0)))
+        f.write_str(&format!("TaskToken({})", format_task_token(&self.0)))
     }
 }
 
@@ -52,6 +52,6 @@ impl Borrow<[u8]> for TaskToken {
     }
 }
 
-pub(crate) fn fmt_tt(tt: &[u8]) -> String {
+pub(crate) fn format_task_token(tt: &[u8]) -> String {
     BASE64_STANDARD.encode(tt)
 }