Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions crates/common/protos/local/temporal/sdk/core/core_interface.proto
Original file line number Diff line number Diff line change
Expand Up @@ -51,3 +51,18 @@ message NexusSlotInfo {
string service = 1;
string operation = 2;
}

// Info about a namespace
message NamespaceInfo {
// Namespace configured limits
Limits limits = 1;

message Limits {
// Maximum size in bytes for payload fields in workflow history events
// (e.g., workflow/activity inputs and results, failure details, signal payloads).
// When exceeded, the server will reject the operation with an error.
int64 blob_size_limit_error = 1;
// Maximum total memo size in bytes per workflow execution.
int64 memo_size_limit_error = 2;
}
}
6 changes: 3 additions & 3 deletions crates/common/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ use crate::{
WorkerValidationError,
},
protos::coresdk::{
ActivityHeartbeat, ActivityTaskCompletion,
ActivityHeartbeat, ActivityTaskCompletion, NamespaceInfo,
activity_task::ActivityTask,
nexus::{NexusTask, NexusTaskCompletion},
workflow_activation::WorkflowActivation,
Expand All @@ -35,7 +35,7 @@ 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>;
async fn validate(&self) -> Result<NamespaceInfo, 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
Expand Down Expand Up @@ -156,7 +156,7 @@ impl<W> Worker for Arc<W>
where
W: Worker + ?Sized,
{
async fn validate(&self) -> Result<(), WorkerValidationError> {
async fn validate(&self) -> Result<NamespaceInfo, WorkerValidationError> {
(**self).validate().await
}

Expand Down
45 changes: 27 additions & 18 deletions crates/sdk-core/src/worker/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,10 @@ use temporalio_common::{
protos::{
TaskToken,
coresdk::{
ActivityTaskCompletion,
ActivityTaskCompletion, NamespaceInfo,
activity_result::activity_execution_result,
activity_task::ActivityTask,
namespace_info,
nexus::{NexusTask, NexusTaskCompletion, nexus_task_completion},
workflow_activation::{WorkflowActivation, remove_from_cache::EvictionReason},
workflow_completion::WorkflowActivationCompletion,
Expand Down Expand Up @@ -164,9 +165,31 @@ pub(crate) struct WorkerTelemetry {

#[async_trait::async_trait]
impl WorkerTrait for Worker {
async fn validate(&self) -> Result<(), WorkerValidationError> {
self.verify_namespace_exists().await?;
Ok(())
async fn validate(&self) -> Result<NamespaceInfo, WorkerValidationError> {
match self.client.describe_namespace().await {
Ok(info) => {
let limits = info.namespace_info.and_then(|ns_info| {
ns_info.limits.map(|api_limits| namespace_info::Limits {
blob_size_limit_error: api_limits.blob_size_limit_error,
memo_size_limit_error: api_limits.memo_size_limit_error,
})
});
return Ok(NamespaceInfo { limits });
}
Err(e) => {
if e.code() == tonic::Code::Unimplemented {
// Ignore if unimplemented since we wouldn't want to fail against an old server, for
// example.
return Ok(NamespaceInfo {
..Default::default()
});
}
return Err(WorkerValidationError::NamespaceDescribeError {
source: e,
namespace: self.config.namespace.clone(),
});
}
}
}

async fn poll_workflow_activation(&self) -> Result<WorkflowActivation, PollError> {
Expand Down Expand Up @@ -1086,20 +1109,6 @@ impl Worker {
dbg_panic!("trying to notify local result when workflows not enabled for this worker");
}
}

async fn verify_namespace_exists(&self) -> Result<(), WorkerValidationError> {
if let Err(e) = self.client.describe_namespace().await {
// Ignore if unimplemented since we wouldn't want to fail against an old server, for
// example.
if e.code() != tonic::Code::Unimplemented {
return Err(WorkerValidationError::NamespaceDescribeError {
source: e,
namespace: self.config.namespace.clone(),
});
}
}
Ok(())
}
}

struct ClientWorkerRegistrator {
Expand Down
Loading