Ensure namespace exists on worker startup

core-api/src/errors.rs

@@ -2,6 +2,17 @@
 
 use temporal_sdk_core_protos::coresdk::activity_result::ActivityExecutionResult;
 
+/// Errors thrown by [crate::Worker::validate]
+#[derive(thiserror::Error, Debug)]
+pub enum WorkerValidationError {
+    /// The namespace provided to the worker does not exist on the server.
+    #[error("Namespace {namespace} was not found or otherwise could not be described: {source:?}")]
+    NamespaceDescribeError {
+        source: tonic::Status,
+        namespace: String,
+    },
+}
+
 /// Errors thrown by [crate::Worker::poll_workflow_activation]
 #[derive(thiserror::Error, Debug)]
 pub enum PollWfError {

core-api/src/lib.rs

@@ -3,7 +3,10 @@ pub mod telemetry;
 pub mod worker;
 
 use crate::{
-    errors::{CompleteActivityError, CompleteWfError, PollActivityError, PollWfError},
+    errors::{
+        CompleteActivityError, CompleteWfError, PollActivityError, PollWfError,
+        WorkerValidationError,
+    },
     worker::WorkerConfig,
 };
 use temporal_sdk_core_protos::coresdk::{
@@ -16,6 +19,11 @@ use temporal_sdk_core_protos::coresdk::{
 /// 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.

core/Cargo.toml

@@ -71,7 +71,7 @@ tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["parking_lot", "env-filter", "registry"] }
 url = "2.2"
 uuid = { version = "1.1", features = ["v4"] }
-zip = { version = "1.2", optional = true }
+zip = { version = "2.0", optional = true }
 
 # 1st party local deps
 [dependencies.temporal-sdk-core-api]

core/src/lib.rs

@@ -51,6 +51,7 @@ use crate::{
     },
     worker::client::WorkerClientBag,
 };
+use anyhow::bail;
 use futures::Stream;
 use std::sync::Arc;
 use temporal_client::{ConfiguredClient, TemporalServiceClientWithMetrics};
@@ -82,7 +83,10 @@ where
 {
     let client = init_worker_client(&worker_config, *client.into().into_inner());
     if client.namespace() != worker_config.namespace {
-        panic!("Passed in client is not bound to the same namespace as the worker");
+        bail!("Passed in client is not bound to the same namespace as the worker");
+    }
+    if client.namespace() == "" {
+        bail!("Namespace cannot be empty");
     }
     let client_ident = client.get_options().identity.clone();
     let sticky_q = sticky_q_name_for_worker(&client_ident, &worker_config);

core/src/worker/client.rs

@@ -3,7 +3,7 @@
 pub(crate) mod mocks;
 use parking_lot::RwLock;
 use std::sync::Arc;
-use temporal_client::{Client, RetryClient, SlotManager, WorkflowService};
+use temporal_client::{Client, Namespace, RetryClient, SlotManager, WorkflowService};
 use temporal_sdk_core_protos::{
     coresdk::workflow_commands::QueryResult,
     temporal::api::{
@@ -145,9 +145,10 @@ pub(crate) trait WorkerClient: Sync + Send {
         task_token: TaskToken,
         query_result: QueryResult,
     ) -> Result<RespondQueryTaskCompletedResponse>;
+    async fn describe_namespace(&self) -> Result<DescribeNamespaceResponse>;
 
     fn replace_client(&self, new_client: RetryClient<Client>);
-    fn capabilities(&self) -> Option<get_system_info_response::Capabilities>;
+    fn capabilities(&self) -> Option<Capabilities>;
     fn workers(&self) -> Arc<SlotManager>;
     fn is_mock(&self) -> bool;
 }
@@ -375,6 +376,14 @@ impl WorkerClient for WorkerClientBag {
             .into_inner())
     }
 
+    async fn describe_namespace(&self) -> Result<DescribeNamespaceResponse> {
+        temporal_client::WorkflowClientTrait::describe_namespace(
+            &self.cloned_client(),
+            Namespace::Name(self.namespace.clone()),
+        )
+        .await
+    }
+
     fn replace_client(&self, new_client: RetryClient<Client>) {
         let mut replaceable_client = self.replaceable_client.write();
         *replaceable_client = new_client;

core/src/worker/client/mocks.rs

@@ -115,12 +115,13 @@ mockall::mock! {
         ) -> impl Future<Output = Result<RespondQueryTaskCompletedResponse>> + Send + 'b
             where 'a: 'b, Self: 'b;
 
-        fn replace_client(&self, new_client: RetryClient<Client>);
-
-        fn capabilities(&self) -> Option<get_system_info_response::Capabilities>;
+        fn describe_namespace<'a, 'b>(&self) ->
+          impl Future<Output = Result<DescribeNamespaceResponse>> + Send + 'b
+          where 'a: 'b, Self: 'b;
 
+        fn replace_client(&self, new_client: RetryClient<Client>);
+        fn capabilities(&self) -> Option<Capabilities>;
         fn workers(&self) -> Arc<SlotManager>;
-
         fn is_mock(&self) -> bool;
     }
 }

core/src/worker/mod.rs

@@ -16,8 +16,6 @@ pub(crate) use activities::{
 };
 pub(crate) use workflow::{wft_poller::new_wft_poller, LEGACY_QUERY_ID};
 
-use temporal_client::{ConfiguredClient, TemporalServiceClientWithMetrics, WorkerKey};
-
 use crate::{
     abstractions::{dbg_panic, MeteredPermitDealer},
     errors::CompleteWfError,
@@ -51,6 +49,7 @@ use std::{
         Arc,
     },
 };
+use temporal_client::{ConfiguredClient, TemporalServiceClientWithMetrics, WorkerKey};
 use temporal_sdk_core_protos::{
     coresdk::{
         activity_result::activity_execution_result,
@@ -70,6 +69,7 @@ use tokio::sync::mpsc::unbounded_channel;
 use tokio_stream::wrappers::UnboundedReceiverStream;
 use tokio_util::sync::CancellationToken;
 
+use temporal_sdk_core_api::errors::WorkerValidationError;
 #[cfg(test)]
 use {
     crate::{
@@ -83,7 +83,7 @@ use {
 /// A worker polls on a certain task queue
 pub struct Worker {
     config: WorkerConfig,
-    wf_client: Arc<dyn WorkerClient>,
+    client: Arc<dyn WorkerClient>,
     /// Registration key to enable eager workflow start for this worker
     worker_key: Mutex<Option<WorkerKey>>,
     /// Manages all workflows and WFT processing
@@ -105,6 +105,11 @@ pub struct Worker {
 
 #[async_trait::async_trait]
 impl WorkerTrait for Worker {
+    async fn validate(&self) -> Result<(), WorkerValidationError> {
+        self.verify_namespace_exists().await?;
+        Ok(())
+    }
+
     async fn poll_workflow_activation(&self) -> Result<WorkflowActivation, PollWfError> {
         self.next_workflow_activation().await
     }
@@ -174,7 +179,7 @@ impl WorkerTrait for Worker {
         self.shutdown_token.cancel();
         // First, disable Eager Workflow Start
         if let Some(key) = *self.worker_key.lock() {
-            self.wf_client.workers().unregister(key);
+            self.client.workers().unregister(key);
         }
         // Second, we want to stop polling of both activity and workflow tasks
         if let Some(atm) = self.at_task_mgr.as_ref() {
@@ -226,11 +231,11 @@ impl Worker {
     pub fn replace_client(&self, new_client: ConfiguredClient<TemporalServiceClientWithMetrics>) {
         // Unregister worker from current client, register in new client at the end
         let mut worker_key = self.worker_key.lock();
-        let slot_provider = (*worker_key).and_then(|k| self.wf_client.workers().unregister(k));
-        self.wf_client
+        let slot_provider = (*worker_key).and_then(|k| self.client.workers().unregister(k));
+        self.client
             .replace_client(super::init_worker_client(&self.config, new_client));
-        *worker_key = slot_provider
-            .and_then(|slot_provider| self.wf_client.workers().register(slot_provider));
+        *worker_key =
+            slot_provider.and_then(|slot_provider| self.client.workers().register(slot_provider));
     }
 
     #[cfg(test)]
@@ -410,7 +415,7 @@ impl Worker {
         let worker_key = Mutex::new(client.workers().register(Box::new(provider)));
         Self {
             worker_key,
-            wf_client: client.clone(),
+            client: client.clone(),
             workflows: Workflows::new(
                 build_wf_basics(
                     config.clone(),
@@ -607,7 +612,7 @@ impl Worker {
         }
 
         if let Some(atm) = &self.at_task_mgr {
-            atm.complete(task_token, status, &*self.wf_client).await;
+            atm.complete(task_token, status, &*self.client).await;
         } else {
             error!(
                 "Tried to complete activity {} on a worker that does not have an activity manager",
@@ -697,6 +702,16 @@ impl Worker {
     fn notify_local_result(&self, run_id: &str, res: LocalResolution) {
         self.workflows.notify_of_local_result(run_id, res);
     }
+
+    async fn verify_namespace_exists(&self) -> Result<(), WorkerValidationError> {
+        if let Err(e) = self.client.describe_namespace().await {
+            return Err(WorkerValidationError::NamespaceDescribeError {
+                source: e,
+                namespace: self.config.namespace.clone(),
+            });
+        }
+        Ok(())
+    }
 }
 
 pub(crate) struct PostActivateHookData<'a> {

tests/integ_tests/worker_tests.rs

@@ -0,0 +1,32 @@
+use assert_matches::assert_matches;
+use temporal_sdk_core::{init_worker, CoreRuntime};
+use temporal_sdk_core_api::{errors::WorkerValidationError, worker::WorkerConfigBuilder, Worker};
+use temporal_sdk_core_test_utils::{get_integ_server_options, get_integ_telem_options};
+
+#[tokio::test]
+async fn worker_validation_fails_on_nonexistent_namespace() {
+    let opts = get_integ_server_options();
+    let runtime = CoreRuntime::new_assume_tokio(get_integ_telem_options()).unwrap();
+    let retrying_client = opts
+        .connect_no_namespace(runtime.telemetry().get_temporal_metric_meter())
+        .await
+        .unwrap();
+
+    let worker = init_worker(
+        &runtime,
+        WorkerConfigBuilder::default()
+            .namespace("i_dont_exist")
+            .task_queue("Wheee!")
+            .worker_build_id("blah")
+            .build()
+            .unwrap(),
+        retrying_client,
+    )
+    .unwrap();
+
+    let res = worker.validate().await;
+    assert_matches!(
+        res,
+        Err(WorkerValidationError::NamespaceDescribeError { .. })
+    );
+}

tests/main.rs

@@ -16,6 +16,7 @@ mod integ_tests {
     mod queries_tests;
     mod update_tests;
     mod visibility_tests;
+    mod worker_tests;
     mod workflow_tests;
 
     use std::str::FromStr;