From df17fbe2194e079bebaf26e2d38f5ee02d4715b7 Mon Sep 17 00:00:00 2001 From: jmaeagle99 <44687433+jmaeagle99@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:44:45 -0700 Subject: [PATCH 1/3] refactor: separate storage driver errors from external storage errors --- crates/common-wasm/src/external_storage.rs | 124 ++++++++---------- crates/common/src/data_converters.rs | 54 ++++++++ .../common/src/external_storage_references.rs | 10 +- crates/common/src/external_storage_runner.rs | 102 +++++++++++--- crates/common/src/lib.rs | 4 +- 5 files changed, 202 insertions(+), 92 deletions(-) create mode 100644 crates/common/src/data_converters.rs diff --git a/crates/common-wasm/src/external_storage.rs b/crates/common-wasm/src/external_storage.rs index 00bd9e2b8..8231f65ba 100644 --- a/crates/common-wasm/src/external_storage.rs +++ b/crates/common-wasm/src/external_storage.rs @@ -111,25 +111,31 @@ pub trait StorageDriverLimiter: Send + Sync { fn acquire<'a>( &'a self, item: &'a Item, - ) -> BoxFuture<'a, Result>; + ) -> BoxFuture<'a, Result>; } const DEFAULT_MAX_DRIVER_OPERATIONS: usize = 64; const DEFAULT_MAX_OPERATIONS_PER_MESSAGE: usize = 8; -fn check_limit(value: usize, name: &str) -> Result<(), ExternalStorageError> { +fn check_limit(value: usize, name: &str) -> Result<(), ExternalStorageConfigError> { if value == 0 { - return Err(ExternalStorageError::InvalidConfiguration(format!( + return Err(ExternalStorageConfigError(format!( "{name} must be at least 1" ))); } Ok(()) } +/// The external storage configuration is not usable. +#[derive(Debug, thiserror::Error)] +#[error("Invalid external storage configuration: {0}")] +pub struct ExternalStorageConfigError(String); + /// Limits on how many external storage requests may be in flight at once. +// Crate-private, along with the methods that set and read it, until the SDK enforces these limits. #[derive(Clone, Debug, PartialEq, Eq)] #[non_exhaustive] -pub struct ExternalStorageConcurrency { +pub(crate) struct ExternalStorageConcurrency { /// Maximum requests in flight at once across every driver registered on the owning /// [`ExternalStorage`]. Each worker and each client gets its own concurrency limit of this /// size. Defaults to 64. @@ -153,11 +159,12 @@ impl Default for ExternalStorageConcurrency { #[bon::bon] impl ExternalStorageConcurrency { /// Create concurrency limits. Each must be at least 1, and each defaults when omitted. - #[builder(state_mod(vis = "pub"))] - pub fn new( + #[allow(dead_code)] + #[builder] + pub(crate) fn new( #[builder(default = DEFAULT_MAX_DRIVER_OPERATIONS)] max_driver_operations: usize, #[builder(default = DEFAULT_MAX_OPERATIONS_PER_MESSAGE)] max_operations_per_message: usize, - ) -> Result { + ) -> Result { check_limit(max_driver_operations, "max_driver_operations")?; check_limit(max_operations_per_message, "max_operations_per_message")?; Ok(Self { @@ -215,6 +222,32 @@ impl fmt::Debug for StorageDriverRetrieveContext { } } +/// Returned as errors from storage drivers. +#[derive(Debug)] +#[non_exhaustive] +pub enum StorageDriverError { + /// Any driver failure. + Other(anyhow::Error), +} + +impl StorageDriverError { + /// The error the driver reported. + pub fn into_source(self) -> anyhow::Error { + match self { + Self::Other(source) => source, + } + } +} + +impl From for StorageDriverError +where + E: Into, +{ + fn from(source: E) -> Self { + Self::Other(source.into()) + } +} + /// Stores and retrieves payloads in an external storage system. /// /// Implementations are called concurrently and must be thread safe. @@ -235,7 +268,7 @@ pub trait StorageDriver: Send + Sync { &self, context: &StorageDriverStoreContext, payloads: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>>; + ) -> BoxFuture<'static, Result, StorageDriverError>>; /// Retrieve the payloads for the given claims, returning one payload per claim in the same /// order. @@ -243,7 +276,7 @@ pub trait StorageDriver: Send + Sync { &self, context: &StorageDriverRetrieveContext, claims: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>>; + ) -> BoxFuture<'static, Result, StorageDriverError>>; } /// Chooses which driver stores a payload, or returns `None` to pass the payload through. @@ -256,49 +289,6 @@ pub trait StorageDriverSelector: Send + Sync { ) -> Option>; } -/// Failures raised by the external storage machinery. -#[derive(Debug, thiserror::Error)] -#[non_exhaustive] -pub enum ExternalStorageError { - /// A reference to externally stored data was found, but no external storage is configured to - /// retrieve it. - #[error( - "[TMPRL1105] Encountered a reference to a payload in external storage, but no external \ - storage is configured to retrieve it" - )] - NotConfigured, - /// A payload is marked as a reference but its contents cannot be read as one. - #[error("Invalid external storage reference: {0}")] - InvalidReference(#[source] Box), - /// The configuration is not usable. - #[error("Invalid external storage configuration: {0}")] - InvalidConfiguration(String), - /// A driver failed to store or retrieve. - #[error("Storage driver error: {0}")] - Driver(#[source] Box), - /// Walking the message's payloads failed, outside any storage operation. - #[error("External storage payload walk failed: {0}")] - Walk(#[source] Box), - /// A reference names a driver that is not registered. - #[error("No storage driver is registered under the name '{0}'")] - UnregisteredDriver(String), - /// The selector returned a driver that is not one of the registered instances. - #[error("Selector returned a driver that is not registered: '{0}'")] - UnregisteredSelection(String), - /// A driver returned a different number of results than it was given inputs. A store is given - /// payloads and returns claims; a retrieve is given claims and returns payloads. Either way - /// there must be one result per input, in order. - #[error("Storage driver '{driver}' returned {returned} results for {requested} inputs")] - DriverContract { - /// Name of the driver. - driver: String, - /// How many inputs it was given. - requested: usize, - /// How many results it returned. - returned: usize, - }, -} - /// Configuration for offloading large payloads to an external storage system. /// /// This and the driver and selector traits it holds are defined in this crate only so that it can @@ -330,7 +320,7 @@ impl fmt::Debug for ExternalStorage { impl ExternalStorage { /// Configure external storage with a single driver that stores every eligible payload. - pub fn new(driver: Arc) -> Result { + pub fn new(driver: Arc) -> Result { let selected = Arc::clone(&driver); Self::with_selector(vec![driver], Arc::new(SingleDriverSelector(selected))) } @@ -342,9 +332,9 @@ impl ExternalStorage { pub fn with_selector( drivers: Vec>, driver_selector: Arc, - ) -> Result { + ) -> Result { if drivers.is_empty() { - return Err(ExternalStorageError::InvalidConfiguration( + return Err(ExternalStorageConfigError( "at least one driver is required".to_owned(), )); } @@ -354,12 +344,12 @@ impl ExternalStorage { // The name is the routing key written into history, so a driver without one can never // be resolved on the retrieval side. if name.is_empty() { - return Err(ExternalStorageError::InvalidConfiguration( + return Err(ExternalStorageConfigError( "driver name cannot be empty".to_owned(), )); } if !names.insert(name) { - return Err(ExternalStorageError::InvalidConfiguration(format!( + return Err(ExternalStorageConfigError(format!( "multiple drivers given with name '{name}'" ))); } @@ -382,14 +372,16 @@ impl ExternalStorage { } /// Set the limits on how many storage requests may be in flight at once. + #[allow(dead_code)] #[must_use] - pub fn with_concurrency(mut self, concurrency: ExternalStorageConcurrency) -> Self { + pub(crate) fn with_concurrency(mut self, concurrency: ExternalStorageConcurrency) -> Self { self.concurrency = concurrency; self } /// The limits on how many storage requests may be in flight at once. - pub fn concurrency(&self) -> &ExternalStorageConcurrency { + #[allow(dead_code)] + pub(crate) fn concurrency(&self) -> &ExternalStorageConcurrency { &self.concurrency } @@ -444,14 +436,14 @@ mod tests { &self, _context: &StorageDriverStoreContext, _payloads: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>> { + ) -> BoxFuture<'static, Result, StorageDriverError>> { Box::pin(future::ready(Ok(vec![]))) } fn retrieve( &self, _context: &StorageDriverRetrieveContext, _claims: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>> { + ) -> BoxFuture<'static, Result, StorageDriverError>> { Box::pin(future::ready(Ok(vec![]))) } } @@ -507,13 +499,13 @@ mod tests { ExternalStorageConcurrency::builder() .max_driver_operations(0) .build(), - Err(ExternalStorageError::InvalidConfiguration(_)) + Err(ExternalStorageConfigError(_)) )); assert!(matches!( ExternalStorageConcurrency::builder() .max_operations_per_message(0) .build(), - Err(ExternalStorageError::InvalidConfiguration(_)) + Err(ExternalStorageConfigError(_)) )); } @@ -521,18 +513,18 @@ mod tests { fn configuration_is_validated_when_built() { assert!(matches!( ExternalStorage::with_selector(vec![], Arc::new(NullSelector)), - Err(ExternalStorageError::InvalidConfiguration(_)) + Err(ExternalStorageConfigError(_)) )); assert!(matches!( ExternalStorage::with_selector(vec![driver("")], Arc::new(NullSelector)), - Err(ExternalStorageError::InvalidConfiguration(_)) + Err(ExternalStorageConfigError(_)) )); assert!(matches!( ExternalStorage::with_selector( vec![driver("dup"), driver("dup")], Arc::new(NullSelector) ), - Err(ExternalStorageError::InvalidConfiguration(_)) + Err(ExternalStorageConfigError(_)) )); } } diff --git a/crates/common/src/data_converters.rs b/crates/common/src/data_converters.rs new file mode 100644 index 000000000..66b44c454 --- /dev/null +++ b/crates/common/src/data_converters.rs @@ -0,0 +1,54 @@ +//! Contains traits for and default implementations of data converters, codecs, and other +//! serialization related functionality. + +// A module of its own rather than a re-export of the one in `temporalio-common-wasm`, so that the +// external storage error, which this crate raises, is published beside the driver contracts it +// belongs with. +pub use temporalio_common_wasm::data_converters::*; + +/// Failures raised while storing payloads in, or retrieving them from, external storage. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum ExternalStorageError { + /// A reference to externally stored data was found, but no external storage is configured to + /// retrieve it. + #[error( + "[TMPRL1105] Encountered a reference to a payload in external storage, but no external \ + storage is configured to retrieve it" + )] + NotConfigured, + /// A payload is marked as a reference but its contents cannot be read as one. + #[error("Invalid external storage reference: {0}")] + InvalidReference(#[source] Box), + /// A driver failed to store or retrieve. + #[error("Storage driver error: {0}")] + Driver(#[source] anyhow::Error), + /// Walking the message's payloads failed, outside any storage operation. + #[error("External storage payload walk failed: {0}")] + Walk(#[source] Box), + /// A reference names a driver that is not registered. + #[error("No storage driver is registered under the name '{0}'")] + UnregisteredDriver(String), + /// The selector returned a driver that is not one of the registered instances. + #[error("Selector returned a driver that is not registered: '{0}'")] + UnregisteredSelection(String), + /// A driver returned a different number of results than it was given inputs. A store is given + /// payloads and returns claims; a retrieve is given claims and returns payloads. Either way + /// there must be one result per input, in order. + #[error("Storage driver '{driver}' returned {returned} results for {requested} inputs")] + DriverContract { + /// Name of the driver. + driver: String, + /// How many inputs it was given. + requested: usize, + /// How many results it returned. + returned: usize, + }, +} + +#[cfg(feature = "experimental")] +impl From for ExternalStorageError { + fn from(error: StorageDriverError) -> Self { + Self::Driver(error.into_source()) + } +} diff --git a/crates/common/src/external_storage_references.rs b/crates/common/src/external_storage_references.rs index 4d5c89943..489c5cb53 100644 --- a/crates/common/src/external_storage_references.rs +++ b/crates/common/src/external_storage_references.rs @@ -11,11 +11,13 @@ use std::collections::HashMap; -use crate::protos::temporal::api::{ - common::v1::{Payload, payload::ExternalPayloadDetails}, - sdk::v1::ExternalStorageReference, +use crate::{ + data_converters::ExternalStorageError, + protos::temporal::api::{ + common::v1::{Payload, payload::ExternalPayloadDetails}, + sdk::v1::ExternalStorageReference, + }, }; -use temporalio_common_wasm::data_converters::ExternalStorageError; const REFERENCE_ENCODING: &str = "json/protobuf"; // Spelled out rather than taken from `prost::Name` so that a proto-side rename fails diff --git a/crates/common/src/external_storage_runner.rs b/crates/common/src/external_storage_runner.rs index 6c0710068..79d328cb4 100644 --- a/crates/common/src/external_storage_runner.rs +++ b/crates/common/src/external_storage_runner.rs @@ -18,12 +18,13 @@ use std::{collections::HashMap, sync::Arc}; use prost::Message; use temporalio_common_wasm::data_converters::{ - ExternalStorage, ExternalStorageError, PayloadConversionError, StorageDriver, - StorageDriverClaim, StorageDriverLimiter, StorageDriverRetrieveContext, - StorageDriverSelectContext, StorageDriverStoreContext, StorageDriverTargetInfo, + ExternalStorage, PayloadConversionError, StorageDriver, StorageDriverClaim, + StorageDriverLimiter, StorageDriverRetrieveContext, StorageDriverSelectContext, + StorageDriverStoreContext, StorageDriverTargetInfo, }; use crate::{ + data_converters::ExternalStorageError, external_storage_references::{create_reference_payload, is_plain_reference, parse_reference}, payload_visitor::{AsyncPayloadVisitor, PayloadVisitable, payload_visitor}, protos::temporal::api::common::v1::Payload, @@ -133,7 +134,10 @@ async fn retrieve_payloads( .limiter(Arc::clone(limiter)) .build(); async move { - let retrieved = driver.retrieve(&context, claims).await?; + let retrieved = driver + .retrieve(&context, claims) + .await + .map_err(ExternalStorageError::from)?; if retrieved.len() != indices.len() { return Err(ExternalStorageError::DriverContract { driver: driver_name, @@ -249,7 +253,10 @@ async fn store_payloads( .limiter(Arc::clone(limiter)) .build(); async move { - let claims = driver.store(&context, to_store).await?; + let claims = driver + .store(&context, to_store) + .await + .map_err(ExternalStorageError::from)?; if claims.len() != group.len() { return Err(ExternalStorageError::DriverContract { driver: driver_name, @@ -285,10 +292,11 @@ mod tests { workflowservice::v1::SignalWithStartWorkflowExecutionRequest, }, }; + use anyhow::anyhow; use futures::future::{self, BoxFuture}; use std::sync::Mutex; use temporalio_common_wasm::data_converters::{ - StorageDriver, StorageDriverPermit, StorageDriverSelector, + StorageDriver, StorageDriverError, StorageDriverPermit, StorageDriverSelector, }; /// Echoes each claim back as a payload (and each payload as a claim), recording the batch @@ -323,7 +331,7 @@ mod tests { &self, _context: &temporalio_common_wasm::data_converters::StorageDriverStoreContext, payloads: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>> { + ) -> BoxFuture<'static, Result, StorageDriverError>> { self.batches.lock().unwrap().push(payloads.len()); let barrier = Arc::clone(&self.barrier); Box::pin(async move { @@ -345,7 +353,7 @@ mod tests { &self, _context: &StorageDriverRetrieveContext, claims: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>> { + ) -> BoxFuture<'static, Result, StorageDriverError>> { self.batches.lock().unwrap().push(claims.len()); let barrier = Arc::clone(&self.barrier); Box::pin(async move { @@ -369,7 +377,7 @@ mod tests { fn acquire<'a>( &'a self, _item: &'a Item, - ) -> BoxFuture<'a, Result> { + ) -> BoxFuture<'a, Result> { Box::pin(future::ready(Ok(StorageDriverPermit::new(Box::new(()))))) } } @@ -481,7 +489,7 @@ mod tests { &self, _context: &StorageDriverStoreContext, payloads: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>> { + ) -> BoxFuture<'static, Result, StorageDriverError>> { let stored = Arc::clone(&self.stored); Box::pin(async move { let mut held = stored.lock().unwrap(); @@ -502,11 +510,43 @@ mod tests { &self, _context: &StorageDriverRetrieveContext, _claims: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>> { + ) -> BoxFuture<'static, Result, StorageDriverError>> { Box::pin(future::ready(Ok(vec![]))) } } + /// Fails every request with an I/O error, converted with `?` as a real driver would. + struct FailingDriver; + + impl FailingDriver { + async fn fail() -> Result { + Err(std::io::Error::other("bucket unreachable"))? + } + } + + impl StorageDriver for FailingDriver { + fn name(&self) -> &str { + "mem" + } + fn driver_type(&self) -> &str { + "test.failing" + } + fn store( + &self, + _context: &StorageDriverStoreContext, + _payloads: Vec, + ) -> BoxFuture<'static, Result, StorageDriverError>> { + Box::pin(Self::fail()) + } + fn retrieve( + &self, + _context: &StorageDriverRetrieveContext, + _claims: Vec, + ) -> BoxFuture<'static, Result, StorageDriverError>> { + Box::pin(Self::fail()) + } + } + /// Returns an empty batch regardless of input, so store and retrieve both violate the /// one-result-per-input contract. struct MiscountingDriver { @@ -523,14 +563,14 @@ mod tests { &self, _context: &StorageDriverStoreContext, _payloads: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>> { + ) -> BoxFuture<'static, Result, StorageDriverError>> { Box::pin(future::ready(Ok(vec![]))) } fn retrieve( &self, _context: &StorageDriverRetrieveContext, _claims: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>> { + ) -> BoxFuture<'static, Result, StorageDriverError>> { Box::pin(future::ready(Ok(vec![]))) } } @@ -682,6 +722,30 @@ mod tests { )); } + #[tokio::test] + async fn a_driver_failure_is_reported_with_the_drivers_error() { + let storage = ExternalStorage::new(Arc::new(FailingDriver)) + .unwrap() + .with_payload_size_threshold(1); + + let store_err = store_payloads(&mut [sized(50)], &storage, None, &store_limiter()) + .await + .unwrap_err(); + let retrieve_err = retrieve_payloads(&mut [reference("mem", "a")], &storage, &limiter()) + .await + .unwrap_err(); + + for err in [store_err, retrieve_err] { + assert!( + matches!(err, ExternalStorageError::Driver(_)), + "got {err:?}" + ); + let source = std::error::Error::source(&err).expect("the driver's error is the source"); + assert!(source.downcast_ref::().is_some()); + assert_eq!(source.to_string(), "bucket unreachable"); + } + } + #[tokio::test] async fn a_retrieve_driver_returning_the_wrong_number_of_payloads_is_an_error() { let storage = ExternalStorage::new(Arc::new(MiscountingDriver { name: "mem" })).unwrap(); @@ -801,7 +865,7 @@ mod tests { &self, _context: &StorageDriverStoreContext, payloads: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>> { + ) -> BoxFuture<'static, Result, StorageDriverError>> { let stored = Arc::clone(&self.stored); Box::pin(async move { let mut held = stored.lock().unwrap(); @@ -819,18 +883,16 @@ mod tests { &self, _context: &StorageDriverRetrieveContext, claims: Vec, - ) -> BoxFuture<'static, Result, ExternalStorageError>> { + ) -> BoxFuture<'static, Result, StorageDriverError>> { let stored = Arc::clone(&self.stored); Box::pin(async move { let held = stored.lock().unwrap(); claims .into_iter() .map(|c| { - held.get(&c.claim_data["key"]).cloned().ok_or_else(|| { - ExternalStorageError::InvalidReference( - "stored payload not found".into(), - ) - }) + held.get(&c.claim_data["key"]) + .cloned() + .ok_or_else(|| anyhow!("stored payload not found").into()) }) .collect() }) diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index dfa784d51..45289ead9 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -7,6 +7,7 @@ #[macro_use] extern crate tracing; +pub mod data_converters; #[cfg(feature = "envconfig")] pub mod envconfig; #[cfg(feature = "experimental")] @@ -25,8 +26,7 @@ pub use temporalio_common_wasm::{ ActivityCloseTimeouts, ActivityDefinition, ActivityError, HasWorkflowDefinition, Memo, MemoValue, MemoValues, Priority, QueryDefinition, RetryPolicy, SignalDefinition, UntypedActivity, UntypedWorkflow, UpdateDefinition, VersioningOverride, - WorkerDeploymentVersion, WorkflowDefinition, WorkflowExecution, data_converters, error, - search_attributes, + WorkerDeploymentVersion, WorkflowDefinition, WorkflowExecution, error, search_attributes, }; macro_rules! dbg_panic { From 8a6fcaecadfa8b4f79fceecbfedb1bc08e0df31c Mon Sep 17 00:00:00 2001 From: jmaeagle99 <44687433+jmaeagle99@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:44:45 -0700 Subject: [PATCH 2/3] feat: reject external storage references with TMPRL1105 --- crates/client/src/activity/activity_handle.rs | 8 ++- crates/client/src/errors.rs | 44 ++++++++++++++- crates/client/src/lib.rs | 13 ++++- crates/client/src/schedules.rs | 26 ++++++--- crates/client/src/workflow_handle.rs | 23 ++++++-- .../common/src/external_storage_references.rs | 56 ++++++++++++++++++- crates/common/src/lib.rs | 1 - crates/common/src/payload_visitor.rs | 21 ++++++- crates/sdk/src/lib.rs | 33 +++++++---- 9 files changed, 188 insertions(+), 37 deletions(-) diff --git a/crates/client/src/activity/activity_handle.rs b/crates/client/src/activity/activity_handle.rs index e8d51af3a..c4292eb33 100644 --- a/crates/client/src/activity/activity_handle.rs +++ b/crates/client/src/activity/activity_handle.rs @@ -10,7 +10,7 @@ use temporalio_common::{ data_converters::{ ActivitySerializationContext, DecodablePayloads, NoopDecodeHint, SerializationContextData, }, - payload_visitor::decode_payloads, + payload_visitor::{decode_payloads, retrieve_external_payloads}, protos::temporal::api::{ activity::v1::{ActivityExecutionOutcome, activity_execution_outcome}, failure::v1::failure::FailureInfo, @@ -79,7 +79,7 @@ where pub async fn result(&self) -> Result { let mut client = self.client.clone(); loop { - let resp = client + let mut resp = client .poll_activity_execution( PollActivityExecutionRequest { namespace: client.namespace(), @@ -90,6 +90,7 @@ where ) .await? .into_inner(); + retrieve_external_payloads(&mut resp, client.data_converter()).await?; // If resp.outcome.value is None, poll again let Some(ActivityExecutionOutcome { @@ -139,7 +140,7 @@ where options: ActivityDescribeOptions, ) -> Result, ActivityInteractionError> { let mut client = self.client.clone(); - let resp = client + let mut resp = client .describe_activity_execution( DescribeActivityExecutionRequest { namespace: client.namespace(), @@ -155,6 +156,7 @@ where ) .await? .into_inner(); + retrieve_external_payloads(&mut resp, client.data_converter()).await?; Ok(ActivityExecutionDescription::new( client.data_converter().clone(), diff --git a/crates/client/src/errors.rs b/crates/client/src/errors.rs index e958511bc..93b12099f 100644 --- a/crates/client/src/errors.rs +++ b/crates/client/src/errors.rs @@ -5,7 +5,7 @@ use crate::PluginApplyError; use crate::{WorkflowExecutionStatus, workflow_handle::WorkflowResultDetails}; use http::uri::InvalidUri; use temporalio_common::{ - data_converters::{DecodablePayloads, PayloadConversionError}, + data_converters::{DecodablePayloads, ExternalStorageError, PayloadConversionError}, error::{IncomingError, TimeoutType}, protos::{ google::rpc::Status as RpcStatus, @@ -114,6 +114,9 @@ pub enum WorkflowStartError { /// Error converting the input to a payload. #[error("Failed to serialize workflow input: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), /// An uncategorized rpc error from the server. #[error("Server error: {0}")] Rpc(#[from] tonic::Status), @@ -154,6 +157,10 @@ pub enum WorkflowQueryError { #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), + /// An uncategorized RPC error from the server. #[error("Server error: {0}")] Rpc(tonic::Status), @@ -189,6 +196,10 @@ pub enum WorkflowUpdateError { #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), + /// An uncategorized RPC error from the server. #[error("Server error: {0}")] Rpc(tonic::Status), @@ -225,6 +236,10 @@ pub enum WorkflowUpdateWithStartError { #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), + /// An RPC error from the server that could not be attributed to either operation. #[error("Server error: {0}")] Rpc(tonic::Status), @@ -319,6 +334,10 @@ pub enum WorkflowGetResultError { #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), + /// An uncategorized RPC error from the server. #[error("Server error: {0}")] Rpc(tonic::Status), @@ -333,6 +352,7 @@ impl From for WorkflowGetResultError { match err { WorkflowInteractionError::NotFound(s) => Self::NotFound(s), WorkflowInteractionError::PayloadConversion(e) => Self::PayloadConversion(e), + WorkflowInteractionError::ExternalStorage(e) => Self::ExternalStorage(e), WorkflowInteractionError::Rpc(s) => Self::Rpc(s), WorkflowInteractionError::Other(e) => Self::Other(e), } @@ -362,6 +382,9 @@ pub enum ClientError { /// Error decoding payloads returned by the server. #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), /// An uncategorized rpc error from the server. #[error("Server error: {0}")] Rpc(#[from] tonic::Status), @@ -380,6 +403,10 @@ pub enum WorkflowInteractionError { #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), + /// An uncategorized RPC error from the server. #[error("Server error: {0}")] Rpc(tonic::Status), @@ -409,6 +436,9 @@ pub enum AsyncActivityError { /// Error serializing an activity result, failure, or details. #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), /// An uncategorized rpc error from the server. #[error("Server error: {0}")] Rpc(#[from] tonic::Status), @@ -446,6 +476,10 @@ pub enum ActivityInteractionError { #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), + /// An uncategorized RPC error from the server. #[error("Server error: {0}")] Rpc(#[source] tonic::Status), @@ -485,6 +519,10 @@ pub enum StartActivityError { #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), + /// An uncategorized RPC error from the server. #[error("Server error: {0}")] Rpc(#[source] tonic::Status), @@ -542,6 +580,10 @@ pub enum ActivityResultError { #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), + /// An uncategorized RPC error from the server. #[error("Server error: {0}")] Rpc(#[source] tonic::Status), diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index d41130b43..1a11a66ac 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -137,7 +137,7 @@ use temporalio_common::{ ActivitySerializationContext, DataConverter, SerializationContext, SerializationContextData, WorkflowSerializationContext, }, - payload_visitor::decode_payloads, + payload_visitor::{decode_payloads, retrieve_external_payloads}, protos::{ coresdk::IntoPayloadsExt, grpc::health::v1::health_client::HealthClient, @@ -2186,11 +2186,13 @@ where let (start_response, update_response) = loop { let mut rpc_request = request.clone().into_request(); rpc_options.apply_to(&mut rpc_request); - let response = + let mut response = WorkflowService::execute_multi_operation(&mut client, rpc_request) .await .map_err(WorkflowUpdateWithStartError::from_status)? .into_inner(); + retrieve_external_payloads(&mut response, client.data_converter()) + .await?; let [start_response, update_response]: [_; 2] = response.responses.try_into().map_err(|_| { @@ -2358,12 +2360,17 @@ where } .into_request(); input.rpc_options.apply_to(&mut request); - let response = WorkflowService::list_workflow_executions( + let mut response = WorkflowService::list_workflow_executions( &mut rpc_client, request, ) .await? .into_inner(); + retrieve_external_payloads( + &mut response, + rpc_client.data_converter(), + ) + .await?; Ok(ListWorkflowsPageOutput::new( response.executions, response.next_page_token, diff --git a/crates/client/src/schedules.rs b/crates/client/src/schedules.rs index 1e58e9f58..11b6a44b7 100644 --- a/crates/client/src/schedules.rs +++ b/crates/client/src/schedules.rs @@ -16,10 +16,10 @@ use std::{ use temporalio_common::{ HasWorkflowDefinition, data_converters::{ - DataConverter, PayloadConversionError, SerializationContextData, TemporalDeserializable, - TemporalSerializable, WorkflowSerializationContext, + DataConverter, ExternalStorageError, PayloadConversionError, SerializationContextData, + TemporalDeserializable, TemporalSerializable, WorkflowSerializationContext, }, - payload_visitor::decode_payloads, + payload_visitor::{decode_payloads, retrieve_external_payloads}, protos::{ coresdk::IntoPayloadsExt, proto_ts_to_system_time, @@ -44,6 +44,9 @@ pub enum ScheduleError { /// Failed to encode workflow input payloads. #[error("Payload conversion error: {0}")] PayloadConversion(#[from] PayloadConversionError), + /// Storing payloads in, or retrieving them from, external storage failed. + #[error("External storage error: {0}")] + ExternalStorage(#[from] ExternalStorageError), /// The server returned a schedule description that is missing required fields. #[error("Malformed schedule description for schedule ID '{schedule_id}': {reason}")] MalformedDescription { @@ -1212,11 +1215,11 @@ where } .into_request(); rpc_options.apply_to(&mut request); - Ok( - WorkflowService::describe_schedule(&mut self.client.clone(), request) - .await? - .into_inner(), - ) + let mut response = WorkflowService::describe_schedule(&mut self.client.clone(), request) + .await? + .into_inner(); + retrieve_external_payloads(&mut response, self.client.data_converter()).await?; + Ok(response) } async fn send_update_raw( @@ -1590,10 +1593,15 @@ where } .into_request(); input.rpc_options.apply_to(&mut request); - let response = + let mut response = WorkflowService::list_schedules(&mut rpc_client, request) .await? .into_inner(); + retrieve_external_payloads( + &mut response, + rpc_client.data_converter(), + ) + .await?; Ok(ListSchedulesPageOutput::new( response.schedules, response.next_page_token, diff --git a/crates/client/src/workflow_handle.rs b/crates/client/src/workflow_handle.rs index 6dcfd7712..3f93c3424 100644 --- a/crates/client/src/workflow_handle.rs +++ b/crates/client/src/workflow_handle.rs @@ -30,7 +30,7 @@ use temporalio_common::{ WorkflowSerializationContext, }, error::IncomingError, - payload_visitor::decode_payloads, + payload_visitor::{decode_payloads, retrieve_external_payloads}, protos::{ coresdk::FromPayloadsExt, proto_ts_to_system_time, @@ -664,7 +664,11 @@ where continue; } - let event_attrs = events.pop().and_then(|ev| ev.attributes); + let mut close_event = events.pop(); + if let Some(event) = close_event.as_mut() { + retrieve_external_payloads(event, self.client.data_converter()).await?; + } + let event_attrs = close_event.and_then(|ev| ev.attributes); macro_rules! follow { ($attrs:ident) => { @@ -882,11 +886,12 @@ where } .into_request(); options.rpc_options.apply_to(&mut request); - let response = client + let mut response = client .query_workflow(request) .await .map_err(WorkflowQueryError::from_status)? .into_inner(); + retrieve_external_payloads(&mut response, &data_converter).await?; Ok(QueryWorkflowOutput::new(response)) }) } @@ -999,7 +1004,7 @@ where options.header, payloads, ); - let response = loop { + let mut response = loop { let mut rpc_request = request.clone().into_request(); options.rpc_options.apply_to(&mut rpc_request); let response = WorkflowService::update_workflow_execution( @@ -1015,6 +1020,8 @@ where break response; } }; + retrieve_external_payloads(&mut response, client.data_converter()) + .await?; let run_id = response .update_ref .as_ref() @@ -1201,11 +1208,13 @@ where } .into_request(); input.options.rpc_options.apply_to(&mut request); - let response = + let mut response = WorkflowService::describe_workflow_execution(&mut client, request) .await .map_err(WorkflowInteractionError::from_status)? .into_inner(); + retrieve_external_payloads(&mut response, client.data_converter()) + .await?; Ok(DescribeWorkflowOutput::new(response)) }) } @@ -1433,13 +1442,15 @@ where } .into_request(); input.rpc_options.apply_to(&mut request); - let response = WorkflowService::poll_workflow_execution_update( + let mut response = WorkflowService::poll_workflow_execution_update( &mut client, request, ) .await .map_err(WorkflowUpdateError::from_status)? .into_inner(); + retrieve_external_payloads(&mut response, client.data_converter()) + .await?; if let Some(outcome) = response.outcome { return Ok(PollWorkflowUpdateOutput::new(outcome)); } diff --git a/crates/common/src/external_storage_references.rs b/crates/common/src/external_storage_references.rs index 489c5cb53..6b8d671ee 100644 --- a/crates/common/src/external_storage_references.rs +++ b/crates/common/src/external_storage_references.rs @@ -6,13 +6,14 @@ //! `temporalio-common-wasm`, because proto JSON support and the payload visitor that will drive it //! are here. -// Crate-private until there is a consumer; nothing offloads or restores payloads yet. +// Only the runner creates or parses references, and nothing calls it yet. #![allow(dead_code)] use std::collections::HashMap; use crate::{ data_converters::ExternalStorageError, + payload_visitor::{PayloadVisitable, payload_visitor}, protos::temporal::api::{ common::v1::{Payload, payload::ExternalPayloadDetails}, sdk::v1::ExternalStorageReference, @@ -38,6 +39,31 @@ pub(crate) fn is_plain_reference(payload: &Payload) -> bool { == Some(REFERENCE_MESSAGE_TYPE.as_bytes()) } +/// Fail if any payload in the message is a reference, for a caller with no external storage to +/// retrieve it with. +pub(crate) async fn reject_references_in( + message: &mut M, +) -> Result<(), ExternalStorageError> { + let mut finder = ReferenceFinder { found: false }; + message + .visit_payloads_mut(&mut finder) + .await + .map_err(|e| ExternalStorageError::Walk(Box::new(e)))?; + if finder.found { + return Err(ExternalStorageError::NotConfigured); + } + Ok(()) +} + +struct ReferenceFinder { + found: bool, +} + +payload_visitor!(ReferenceFinder, |this, payloads| { + this.found |= payloads.iter().any(is_plain_reference); + payloads +}); + /// Parse the reference from the given payload, which must be a plain reference. pub(crate) fn parse_reference( payload: &Payload, @@ -88,6 +114,7 @@ pub(crate) fn create_reference_payload( #[cfg(test)] mod tests { use super::*; + use crate::protos::temporal::api::common::v1::Payloads; #[test] fn reference_payload_uses_the_cross_sdk_wire_shape() { @@ -198,6 +225,33 @@ mod tests { assert!(is_plain_reference(&payload)); } + #[tokio::test] + async fn a_reference_with_no_storage_to_retrieve_it_is_tmprl1105() { + let ordinary = Payload { + data: b"fine".to_vec(), + ..Default::default() + }; + let reference = create_reference_payload( + "mem", + HashMap::from([("key".to_owned(), "a".to_owned())]), + 99, + ) + .unwrap(); + let mut message = Payloads { + payloads: vec![ordinary.clone(), reference], + }; + + let err = reject_references_in(&mut message).await.unwrap_err(); + + assert!(matches!(err, ExternalStorageError::NotConfigured)); + assert!(err.to_string().contains("TMPRL1105")); + reject_references_in(&mut Payloads { + payloads: vec![ordinary], + }) + .await + .unwrap(); + } + #[test] fn ordinary_payload_is_not_a_reference() { let payload = Payload { diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 45289ead9..616c4cc70 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -10,7 +10,6 @@ extern crate tracing; pub mod data_converters; #[cfg(feature = "envconfig")] pub mod envconfig; -#[cfg(feature = "experimental")] pub(crate) mod external_storage_references; #[cfg(feature = "experimental")] pub(crate) mod external_storage_runner; diff --git a/crates/common/src/payload_visitor.rs b/crates/common/src/payload_visitor.rs index 9f3635620..31e27dba2 100644 --- a/crates/common/src/payload_visitor.rs +++ b/crates/common/src/payload_visitor.rs @@ -5,7 +5,10 @@ //! boundary between the SDK and Core. use crate::{ - data_converters::{PayloadCodec, PayloadConversionError, SerializationContextData}, + data_converters::{ + DataConverter, ExternalStorageError, PayloadCodec, PayloadConversionError, + SerializationContextData, + }, protos::temporal::api::common::v1::{Payload, Payloads}, }; use futures::future::BoxFuture; @@ -103,7 +106,6 @@ macro_rules! payload_visitor { } }; } -#[cfg(feature = "experimental")] pub(crate) use payload_visitor; /// Visitor that encodes payloads using a codec. @@ -146,6 +148,21 @@ pub async fn decode_payloads( msg.visit_payloads_mut(&mut visitor).await } +/// Replace every external storage reference in an inbound message with the payload it points at. +/// Runs before [`decode_payloads`], since a driver returns a payload as it was before offloading. +/// Without external storage to retrieve them, a reference is an error instead. +/// +/// On error the message is left partly processed and must not be used. +#[doc(hidden)] +pub async fn retrieve_external_payloads( + msg: &mut M, + data_converter: &DataConverter, +) -> Result<(), ExternalStorageError> { + // External storage cannot be configured yet, so there is never anything to retrieve with. + let _ = data_converter; + crate::external_storage_references::reject_references_in(msg).await +} + // Manual impl for Payload - visits itself as a single payload impl PayloadVisitable for Payload { fn visit_payloads_mut<'a>( diff --git a/crates/sdk/src/lib.rs b/crates/sdk/src/lib.rs index 2e7bc67c0..88f9205ef 100644 --- a/crates/sdk/src/lib.rs +++ b/crates/sdk/src/lib.rs @@ -151,7 +151,7 @@ use temporalio_common::{ ActivitySerializationContext, DataConverter, SerializationContext, SerializationContextData, WorkflowSerializationContext, }, - payload_visitor::{decode_payloads, encode_payloads}, + payload_visitor::{decode_payloads, encode_payloads, retrieve_external_payloads}, protos::{ TaskToken, coresdk::{ @@ -1194,13 +1194,19 @@ impl Worker { source: Box::new(source), })?, }; - if let Err(err) = decode_payloads( - &mut activation, - common.data_converter.codec(), - &SerializationContextData::Workflow(WorkflowSerializationContext::new()), - ) - .await - { + let decoded = async { + retrieve_external_payloads(&mut activation, &common.data_converter) + .await?; + decode_payloads( + &mut activation, + common.data_converter.codec(), + &SerializationContextData::Workflow(WorkflowSerializationContext::new()), + ) + .await?; + Ok::<_, anyhow::Error>(()) + } + .await; + if let Err(err) = decoded { let run_id = activation.run_id; error!(run_id, error = %err, "Failed decoding workflow activation"); completions_tx @@ -1280,7 +1286,9 @@ impl Worker { message: "activity polling failed".to_owned(), source: Box::new(source), })?; - if let Err(err) = + let decoded = async { + retrieve_external_payloads(&mut activity, &common.data_converter) + .await?; decode_payloads( &mut activity, common.data_converter.codec(), @@ -1288,8 +1296,11 @@ impl Worker { ActivitySerializationContext::new(), ), ) - .await - { + .await?; + Ok::<_, anyhow::Error>(()) + } + .await; + if let Err(err) = decoded { error!(error = %err, "Failed decoding activity task"); let mut completion = ActivityTaskCompletion { task_token: activity.task_token, From d1fa6682c4ae876e505597b3765516206f9dbf54 Mon Sep 17 00:00:00 2001 From: jmaeagle99 <44687433+jmaeagle99@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:44:46 -0700 Subject: [PATCH 3/3] feat: offload large payloads via external storage --- changelog/added/hoarding-hamster.md | 1 + crates/client/Cargo.toml | 2 +- crates/client/src/async_activity_handle.rs | 42 +- crates/client/src/lib.rs | 108 ++-- crates/client/src/schedules.rs | 12 +- crates/client/src/workflow_handle.rs | 20 +- crates/common-wasm/src/data_converters.rs | 21 +- crates/common-wasm/src/external_storage.rs | 33 +- .../common/src/external_storage_references.rs | 5 +- crates/common/src/external_storage_runner.rs | 94 ++- crates/common/src/payload_visitor.rs | 35 +- .../integ_tests/external_storage_tests.rs | 582 ++++++++++++++++++ crates/sdk-core/tests/main.rs | 1 + crates/sdk/src/lib.rs | 42 +- 14 files changed, 862 insertions(+), 136 deletions(-) create mode 100644 changelog/added/hoarding-hamster.md create mode 100644 crates/sdk-core/tests/integ_tests/external_storage_tests.rs diff --git a/changelog/added/hoarding-hamster.md b/changelog/added/hoarding-hamster.md new file mode 100644 index 000000000..83c9b0c07 --- /dev/null +++ b/changelog/added/hoarding-hamster.md @@ -0,0 +1 @@ +Experimental external storage (requires the `experimental` feature). `DataConverter::with_external_storage` offloads payloads at or above a size threshold to a user-supplied `StorageDriver`, replacing them in history with small references that workers and clients resolve transparently. See `ExternalStorage` for current limitations. This API may change without notice. diff --git a/crates/client/Cargo.toml b/crates/client/Cargo.toml index a69582b5b..1ab16b26e 100644 --- a/crates/client/Cargo.toml +++ b/crates/client/Cargo.toml @@ -17,7 +17,7 @@ features = ["experimental"] [features] default = ["envconfig", "tls-ring"] -experimental = [] +experimental = ["temporalio-common/experimental"] tls-ring = ["tonic/tls-ring"] tls-aws-lc = ["tonic/tls-aws-lc"] telemetry = ["dep:opentelemetry"] diff --git a/crates/client/src/async_activity_handle.rs b/crates/client/src/async_activity_handle.rs index 66b100b59..ed35d8dc3 100644 --- a/crates/client/src/async_activity_handle.rs +++ b/crates/client/src/async_activity_handle.rs @@ -12,7 +12,7 @@ use temporalio_common::{ TemporalSerializable, }, error::{ApplicationFailure, OutgoingActivityError, OutgoingError}, - payload_visitor::encode_payloads, + payload_visitor::{encode_payloads, store_external_payloads}, protos::{ TaskToken, temporal::api::{ @@ -174,8 +174,9 @@ impl AsyncActivityHandle { identity: client.identity(), namespace: client.namespace(), ..Default::default() - } - .into_request(); + }; + store_external_payloads(&mut request, client.data_converter()).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); WorkflowService::respond_activity_task_completed( &mut client, @@ -193,8 +194,9 @@ impl AsyncActivityHandle { result, identity: client.identity(), resource_id: Default::default(), - } - .into_request(); + }; + store_external_payloads(&mut request, client.data_converter()).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); WorkflowService::respond_activity_task_completed_by_id( &mut client, @@ -262,8 +264,9 @@ impl AsyncActivityHandle { namespace: client.namespace(), last_heartbeat_details, ..Default::default() - } - .into_request(); + }; + store_external_payloads(&mut request, client.data_converter()).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); WorkflowService::respond_activity_task_failed( &mut client, @@ -283,8 +286,9 @@ impl AsyncActivityHandle { last_heartbeat_details, resource_id: Default::default(), ..Default::default() - } - .into_request(); + }; + store_external_payloads(&mut request, client.data_converter()).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); WorkflowService::respond_activity_task_failed_by_id( &mut client, @@ -333,8 +337,9 @@ impl AsyncActivityHandle { identity: client.identity(), namespace: client.namespace(), ..Default::default() - } - .into_request(); + }; + store_external_payloads(&mut request, client.data_converter()).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); WorkflowService::respond_activity_task_canceled( &mut client, @@ -352,8 +357,9 @@ impl AsyncActivityHandle { details, identity: client.identity(), ..Default::default() - } - .into_request(); + }; + store_external_payloads(&mut request, client.data_converter()).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); WorkflowService::respond_activity_task_canceled_by_id( &mut client, @@ -401,8 +407,9 @@ impl AsyncActivityHandle { identity: client.identity(), namespace: client.namespace(), resource_id: Default::default(), - } - .into_request(); + }; + store_external_payloads(&mut request, client.data_converter()).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); let response = WorkflowService::record_activity_task_heartbeat( &mut client, @@ -422,8 +429,9 @@ impl AsyncActivityHandle { details, identity: client.identity(), resource_id: Default::default(), - } - .into_request(); + }; + store_external_payloads(&mut request, client.data_converter()).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); let response = WorkflowService::record_activity_task_heartbeat_by_id( diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 1a11a66ac..42e0b6c11 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -137,7 +137,7 @@ use temporalio_common::{ ActivitySerializationContext, DataConverter, SerializationContext, SerializationContextData, WorkflowSerializationContext, }, - payload_visitor::{decode_payloads, retrieve_external_payloads}, + payload_visitor::{decode_payloads, retrieve_external_payloads, store_external_payloads}, protos::{ coresdk::IntoPayloadsExt, grpc::health::v1::health_client::HealthClient, @@ -1906,8 +1906,9 @@ where payloads.into_payloads(), memo, options, - ) - .into_request(); + ); + store_external_payloads(&mut request, &data_converter).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); let run_id = client .start_workflow_execution(request) @@ -2037,8 +2038,9 @@ where priority: start_request.priority, time_skipping_config: start_request.time_skipping_config, ..Default::default() - } - .into_request(); + }; + store_external_payloads(&mut request, &data_converter).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); let run_id = WorkflowService::signal_with_start_workflow_execution( &mut client, @@ -2166,7 +2168,7 @@ where update_payloads, ); - let request = ExecuteMultiOperationRequest { + let mut request = ExecuteMultiOperationRequest { namespace, operations: vec![ execute_multi_operation_request::Operation { @@ -2182,6 +2184,7 @@ where ], resource_id: workflow_id.clone(), }; + store_external_payloads(&mut request, &data_converter).await?; let (start_response, update_response) = loop { let mut rpc_request = request.clone().into_request(); @@ -2504,57 +2507,50 @@ where (summary.is_some() || details.is_some()).then_some(UserMetadata { summary, details }) }; + let mut request = StartActivityExecutionRequest { + namespace: client.namespace(), + identity: client.identity(), + request_id: Uuid::new_v4().to_string(), + activity_id: options.id.clone(), + activity_type: Some(ActivityType { + name: activity.name().to_string(), + }), + task_queue: Some(TaskQueue { + name: options.task_queue, + kind: TaskQueueKind::Normal.into(), + normal_name: "".to_string(), + }), + schedule_to_close_timeout: try_into_or_box_err( + options.close_timeouts.schedule_to_close(), + StartActivityError::Other, + )?, + schedule_to_start_timeout: try_into_or_box_err( + options.schedule_to_start_timeout, + StartActivityError::Other, + )?, + start_to_close_timeout: try_into_or_box_err( + options.close_timeouts.start_to_close(), + StartActivityError::Other, + )?, + heartbeat_timeout: try_into_or_box_err( + options.heartbeat_timeout, + StartActivityError::Other, + )?, + retry_policy: options.retry_policy.map(Into::into), + input: dc.to_payloads(sc, &input).await?.into_payloads(), + id_reuse_policy: ProtoActivityIdReusePolicy::from(options.id_reuse_policy).into(), + id_conflict_policy: ProtoActivityIdConflictPolicy::from(options.id_conflict_policy) + .into(), + search_attributes: options.search_attributes.map(SearchAttributes::into_proto), + header: options.header, + user_metadata, + priority: Some(options.priority.into()), + start_delay: try_into_or_box_err(options.start_delay, StartActivityError::Other)?, + ..Default::default() + }; + store_external_payloads(&mut request, dc).await?; let resp = client - .start_activity_execution( - StartActivityExecutionRequest { - namespace: client.namespace(), - identity: client.identity(), - request_id: Uuid::new_v4().to_string(), - activity_id: options.id.clone(), - activity_type: Some(ActivityType { - name: activity.name().to_string(), - }), - task_queue: Some(TaskQueue { - name: options.task_queue, - kind: TaskQueueKind::Normal.into(), - normal_name: "".to_string(), - }), - schedule_to_close_timeout: try_into_or_box_err( - options.close_timeouts.schedule_to_close(), - StartActivityError::Other, - )?, - schedule_to_start_timeout: try_into_or_box_err( - options.schedule_to_start_timeout, - StartActivityError::Other, - )?, - start_to_close_timeout: try_into_or_box_err( - options.close_timeouts.start_to_close(), - StartActivityError::Other, - )?, - heartbeat_timeout: try_into_or_box_err( - options.heartbeat_timeout, - StartActivityError::Other, - )?, - retry_policy: options.retry_policy.map(Into::into), - input: dc.to_payloads(sc, &input).await?.into_payloads(), - id_reuse_policy: ProtoActivityIdReusePolicy::from(options.id_reuse_policy) - .into(), - id_conflict_policy: ProtoActivityIdConflictPolicy::from( - options.id_conflict_policy, - ) - .into(), - search_attributes: options.search_attributes.map(SearchAttributes::into_proto), - header: options.header, - user_metadata, - priority: Some(options.priority.into()), - start_delay: try_into_or_box_err( - options.start_delay, - StartActivityError::Other, - )?, - ..Default::default() - } - .into_request(), - ) + .start_activity_execution(request.into_request()) .await? .into_inner(); diff --git a/crates/client/src/schedules.rs b/crates/client/src/schedules.rs index 11b6a44b7..c14eaa1ee 100644 --- a/crates/client/src/schedules.rs +++ b/crates/client/src/schedules.rs @@ -19,7 +19,7 @@ use temporalio_common::{ DataConverter, ExternalStorageError, PayloadConversionError, SerializationContextData, TemporalDeserializable, TemporalSerializable, WorkflowSerializationContext, }, - payload_visitor::{decode_payloads, retrieve_external_payloads}, + payload_visitor::{decode_payloads, retrieve_external_payloads, store_external_payloads}, protos::{ coresdk::IntoPayloadsExt, proto_ts_to_system_time, @@ -1238,8 +1238,9 @@ where identity: self.client.identity(), request_id: Uuid::new_v4().to_string(), ..Default::default() - } - .into_request(); + }; + store_external_payloads(&mut request, self.client.data_converter()).await?; + let mut request = request.into_request(); rpc_options.apply_to(&mut request); WorkflowService::update_schedule(&mut self.client.clone(), request).await?; Ok(()) @@ -1517,8 +1518,9 @@ impl Client { identity: client.identity(), request_id: Uuid::new_v4().to_string(), ..Default::default() - } - .into_request(); + }; + store_external_payloads(&mut request, client.data_converter()).await?; + let mut request = request.into_request(); options.rpc_options.apply_to(&mut request); WorkflowService::create_schedule(&mut client, request).await?; Ok(CreateScheduleOutput::new(input.schedule_id)) diff --git a/crates/client/src/workflow_handle.rs b/crates/client/src/workflow_handle.rs index 3f93c3424..3f5f5b21a 100644 --- a/crates/client/src/workflow_handle.rs +++ b/crates/client/src/workflow_handle.rs @@ -30,7 +30,7 @@ use temporalio_common::{ WorkflowSerializationContext, }, error::IncomingError, - payload_visitor::{decode_payloads, retrieve_external_payloads}, + payload_visitor::{decode_payloads, retrieve_external_payloads, store_external_payloads}, protos::{ coresdk::FromPayloadsExt, proto_ts_to_system_time, @@ -810,8 +810,9 @@ where .unwrap_or_else(|| Uuid::new_v4().to_string()), header: options.header, ..Default::default() - } - .into_request(); + }; + store_external_payloads(&mut request, &data_converter).await?; + let mut request = request.into_request(); options.rpc_options.apply_to(&mut request); WorkflowService::signal_workflow_execution(&mut client, request) .await @@ -883,8 +884,9 @@ where .reject_condition .map(|condition| ProtoQueryRejectCondition::from(condition) as i32) .unwrap_or(ProtoQueryRejectCondition::None as i32), - } - .into_request(); + }; + store_external_payloads(&mut request, &data_converter).await?; + let mut request = request.into_request(); options.rpc_options.apply_to(&mut request); let mut response = client .query_workflow(request) @@ -994,7 +996,7 @@ where let update_id = options .update_id .unwrap_or_else(|| Uuid::new_v4().to_string()); - let request = build_update_workflow_request( + let mut request = build_update_workflow_request( client.namespace(), client.identity(), workflow_id.clone(), @@ -1004,6 +1006,7 @@ where options.header, payloads, ); + store_external_payloads(&mut request, &data_converter).await?; let mut response = loop { let mut rpc_request = request.clone().into_request(); options.rpc_options.apply_to(&mut rpc_request); @@ -1163,8 +1166,9 @@ where identity: client.identity(), first_execution_run_id: input.first_execution_run_id, links: vec![], - } - .into_request(); + }; + store_external_payloads(&mut request, client.data_converter()).await?; + let mut request = request.into_request(); input.options.rpc_options.apply_to(&mut request); WorkflowService::terminate_workflow_execution(&mut client, request) .await diff --git a/crates/common-wasm/src/data_converters.rs b/crates/common-wasm/src/data_converters.rs index 5e587ab5b..15714a326 100644 --- a/crates/common-wasm/src/data_converters.rs +++ b/crates/common-wasm/src/data_converters.rs @@ -14,7 +14,6 @@ pub use failure_converter::{ // are defined in this crate: drivers are implemented and invoked outside of WASM, by the worker and // client, never from workflow code. #[cfg(feature = "experimental")] -#[doc(hidden)] pub use crate::external_storage::*; use well_known::{BINARY_NULL_ENCODING_VAL, WellKnownType, binary_null_payload}; @@ -32,6 +31,8 @@ pub struct DataConverter { #[allow(dead_code)] // Will be used for failure conversion failure_converter: Arc, codec: Arc, + #[cfg(feature = "experimental")] + external_storage: Option>, } impl std::fmt::Debug for DataConverter { @@ -53,9 +54,27 @@ impl DataConverter { payload_converter, failure_converter: Arc::new(failure_converter), codec: Arc::new(codec), + #[cfg(feature = "experimental")] + external_storage: None, } } + /// Offload payloads at or above the configured size threshold to external storage, keeping + /// large data out of workflow history. See [`ExternalStorage`] for how offloading works and its + /// current limitations. + #[cfg(feature = "experimental")] + #[must_use] + pub fn with_external_storage(mut self, external_storage: ExternalStorage) -> Self { + self.external_storage = Some(Arc::new(external_storage)); + self + } + + /// Returns the external storage configuration, if any. + #[cfg(feature = "experimental")] + pub fn external_storage(&self) -> Option<&ExternalStorage> { + self.external_storage.as_deref() + } + /// Serialize a value into a single payload, applying the codec. pub async fn to_payload( &self, diff --git a/crates/common-wasm/src/external_storage.rs b/crates/common-wasm/src/external_storage.rs index 8231f65ba..34b6f9b32 100644 --- a/crates/common-wasm/src/external_storage.rs +++ b/crates/common-wasm/src/external_storage.rs @@ -5,8 +5,7 @@ //! [`DataConverter`](crate::data_converters::DataConverter) the configuration hangs off, which is //! the only reason they are defined in this crate: drivers are implemented and invoked outside of //! WASM, by the worker and client, never from workflow code. The reference wire format lives in -//! `temporalio-common`, alongside the payload visitor and the proto JSON support it needs. Nothing -//! here is wired into the SDK yet. +//! `temporalio-common`, alongside the payload visitor and the proto JSON support it needs. use crate::protos::temporal::api::common::v1::Payload; use futures::future::BoxFuture; @@ -179,7 +178,8 @@ impl ExternalStorageConcurrency { #[builder(state_mod(vis = "pub"))] #[non_exhaustive] pub struct StorageDriverStoreContext { - /// The execution the payloads are being stored on behalf of, when there is one. + /// The execution the payloads are being stored on behalf of, when there is one. The SDK does + /// not set this yet, so for now it is always `None`. pub target: Option, /// Limiter for this driver's store requests. pub limiter: Arc>, @@ -198,7 +198,8 @@ impl fmt::Debug for StorageDriverStoreContext { #[builder(state_mod(vis = "pub"))] #[non_exhaustive] pub struct StorageDriverSelectContext { - /// The execution the payload is being stored on behalf of, when there is one. + /// The execution the payload is being stored on behalf of, when there is one. The SDK does not + /// set this yet, so for now it is always `None`. pub target: Option, } @@ -289,14 +290,30 @@ pub trait StorageDriverSelector: Send + Sync { ) -> Option>; } -/// Configuration for offloading large payloads to an external storage system. +/// Configuration for offloading large payloads to an external storage system, so that their data +/// never reaches the Temporal server. /// -/// This and the driver and selector traits it holds are defined in this crate only so that it can -/// be configured on a [`DataConverter`](crate::data_converters::DataConverter). Drivers are -/// implemented and invoked outside of WASM, by the worker and client, never from workflow code. +/// A payload at or above [`Self::payload_size_threshold`] is handed to a [`StorageDriver`] and +/// replaced on the wire by a small reference to what the driver stored. Workers and clients swap +/// references back for the payloads they point at when they read them, except that a workflow's +/// fetched history is returned as stored, references included. Configure it with +/// [`DataConverter::with_external_storage`](crate::data_converters::DataConverter::with_external_storage). +/// Drivers are called by the worker and the client, never from workflow code. /// /// Validated when constructed rather than when used, so a misconfiguration surfaces at startup /// instead of on the first payload large enough to offload. +/// +/// # Limitations +/// +/// This API is experimental and may change without notice. For now: +/// +/// - Heartbeat details recorded from inside an activity are not offloaded. +/// - Drivers are not told which workflow or activity a payload belongs to: the `target` on +/// [`StorageDriverStoreContext`] and [`StorageDriverSelectContext`] is always `None`. +/// - Storage requests are not limited, so every request a message needs runs at once. +/// - References are not passed through the payload codec, so their claim data is written to history +/// unencrypted, and a reference that another SDK encoded with a codec is not recognized. +/// - Replaying a workflow stores its payloads again. #[derive(Clone)] pub struct ExternalStorage { drivers: Vec>, diff --git a/crates/common/src/external_storage_references.rs b/crates/common/src/external_storage_references.rs index 6b8d671ee..699fb5504 100644 --- a/crates/common/src/external_storage_references.rs +++ b/crates/common/src/external_storage_references.rs @@ -6,8 +6,9 @@ //! `temporalio-common-wasm`, because proto JSON support and the payload visitor that will drive it //! are here. -// Only the runner creates or parses references, and nothing calls it yet. -#![allow(dead_code)] +// Only the runner, which needs the `experimental` feature, creates or parses references. Detecting +// them is needed in every build. +#![cfg_attr(not(feature = "experimental"), allow(dead_code))] use std::collections::HashMap; diff --git a/crates/common/src/external_storage_runner.rs b/crates/common/src/external_storage_runner.rs index 79d328cb4..473a9dbbd 100644 --- a/crates/common/src/external_storage_runner.rs +++ b/crates/common/src/external_storage_runner.rs @@ -2,7 +2,7 @@ //! //! Outbound, a payload at or above the configured size threshold is handed to a storage driver and //! replaced by a reference to what the driver stored. Inbound, references are swapped back for the -//! payloads they point at. [`ExternalStorageRunner`] runs either direction over a whole message, +//! payloads they point at. [`store_in`] and [`retrieve_in`] run either direction over a whole message, //! visiting every payload-bearing field except search attributes, which the server has to be able //! to read. //! @@ -10,17 +10,17 @@ //! claim keys) is written to history unencrypted. The payload codec is not applied to references, //! and the configured concurrency limits are not enforced. -// Crate-private until a worker or client calls it. -#![allow(dead_code)] - -use futures::stream::{FuturesUnordered, TryStreamExt}; +use futures::{ + future::{self, BoxFuture}, + stream::{FuturesUnordered, TryStreamExt}, +}; use std::{collections::HashMap, sync::Arc}; use prost::Message; use temporalio_common_wasm::data_converters::{ - ExternalStorage, PayloadConversionError, StorageDriver, StorageDriverClaim, - StorageDriverLimiter, StorageDriverRetrieveContext, StorageDriverSelectContext, - StorageDriverStoreContext, StorageDriverTargetInfo, + ExternalStorage, PayloadConversionError, StorageDriver, StorageDriverClaim, StorageDriverError, + StorageDriverLimiter, StorageDriverPermit, StorageDriverRetrieveContext, + StorageDriverSelectContext, StorageDriverStoreContext, StorageDriverTargetInfo, }; use crate::{ @@ -30,6 +30,55 @@ use crate::{ protos::temporal::api::common::v1::Payload, }; +/// Replace every eligible payload in an outbound message with a reference. Does nothing when no +/// external storage is configured. +/// +/// On error the message is left partly processed and must not be used. +pub(crate) async fn store_in( + storage: Option<&ExternalStorage>, + message: &mut M, + target: Option<&StorageDriverTargetInfo>, +) -> Result<(), ExternalStorageError> { + match storage { + Some(storage) => { + ExternalStorageRunner::new(storage) + .store_in(message, target) + .await + } + None => Ok(()), + } +} + +/// Replace every reference in an inbound message with the payload it points at. With no external +/// storage configured, a reference is an error rather than something to hand to application code +/// as though it were the application's own data. +/// +/// On error the message is left partly processed and must not be used. +pub(crate) async fn retrieve_in( + storage: Option<&ExternalStorage>, + message: &mut M, +) -> Result<(), ExternalStorageError> { + match storage { + Some(storage) => { + ExternalStorageRunner::new(storage) + .retrieve_in(message) + .await + } + None => crate::external_storage_references::reject_references_in(message).await, + } +} + +struct UnboundedLimiter; + +impl StorageDriverLimiter for UnboundedLimiter { + fn acquire<'a>( + &'a self, + _item: &'a Item, + ) -> BoxFuture<'a, Result> { + Box::pin(future::ready(Ok(StorageDriverPermit::new(Box::new(()))))) + } +} + /// Drives external storage for the payloads reachable from one activation or request. /// /// A worker completion holds many commands that offload to different targets, so the caller @@ -42,15 +91,11 @@ pub(crate) struct ExternalStorageRunner<'a> { } impl<'a> ExternalStorageRunner<'a> { - pub(crate) fn new( - storage: &'a ExternalStorage, - store_limiter: Arc>, - retrieve_limiter: Arc>, - ) -> Self { + pub(crate) fn new(storage: &'a ExternalStorage) -> Self { Self { storage, - store_limiter, - retrieve_limiter, + store_limiter: Arc::new(UnboundedLimiter), + retrieve_limiter: Arc::new(UnboundedLimiter), } } @@ -911,7 +956,7 @@ mod tests { ) .unwrap() .with_payload_size_threshold(1); - let runner = ExternalStorageRunner::new(&storage, store_limiter(), limiter()); + let runner = ExternalStorageRunner::new(&storage); let mut request = request_with_large_payloads(); let before = request.clone(); @@ -930,7 +975,7 @@ mod tests { async fn the_runner_reports_a_driver_it_does_not_know() { let batches = Arc::new(Mutex::new(vec![])); let storage = storage("mem", Arc::clone(&batches)); - let runner = ExternalStorageRunner::new(&storage, store_limiter(), limiter()); + let runner = ExternalStorageRunner::new(&storage); // A reference to the registered driver sits beside the unknown one. let mut request = SignalWithStartWorkflowExecutionRequest { input: Some(Payloads { @@ -1001,7 +1046,7 @@ mod tests { stored: Arc>>, ) { let storage = storage_with_threshold(1, stored); - let runner = ExternalStorageRunner::new(&storage, store_limiter(), limiter()); + let runner = ExternalStorageRunner::new(&storage); runner.store_in(request, None).await.unwrap(); } @@ -1032,7 +1077,7 @@ mod tests { async fn every_other_payload_bearing_field_is_retrieved() { let batches = Arc::new(Mutex::new(vec![])); let storage = storage("mem", Arc::clone(&batches)); - let runner = ExternalStorageRunner::new(&storage, store_limiter(), limiter()); + let runner = ExternalStorageRunner::new(&storage); let mut request = request_with_references(); let search_attributes = request.search_attributes.clone(); @@ -1057,4 +1102,15 @@ mod tests { sizes.sort_unstable(); assert_eq!(sizes, vec![1, 1, 1, 2]); } + + #[tokio::test] + async fn with_no_storage_configured_messages_pass_through() { + let mut request = request_with_large_payloads(); + let before = request.clone(); + + store_in(None, &mut request, None).await.unwrap(); + retrieve_in(None, &mut request).await.unwrap(); + + assert_eq!(request, before); + } } diff --git a/crates/common/src/payload_visitor.rs b/crates/common/src/payload_visitor.rs index 31e27dba2..b2367ac31 100644 --- a/crates/common/src/payload_visitor.rs +++ b/crates/common/src/payload_visitor.rs @@ -148,9 +148,31 @@ pub async fn decode_payloads( msg.visit_payloads_mut(&mut visitor).await } +/// Offload every eligible payload in an outbound message to the data converter's external storage. +/// Runs after [`encode_payloads`], since the size threshold applies to encoded payloads. Does +/// nothing without the `experimental` feature. +/// +/// On error the message is left partly processed and must not be sent. +#[doc(hidden)] +pub async fn store_external_payloads( + msg: &mut M, + data_converter: &DataConverter, +) -> Result<(), ExternalStorageError> { + // Stored without a target until the worker and client can name the execution a message is for. + #[cfg(feature = "experimental")] + return crate::external_storage_runner::store_in(data_converter.external_storage(), msg, None) + .await; + #[cfg(not(feature = "experimental"))] + { + let _ = (msg, data_converter); + Ok(()) + } +} + /// Replace every external storage reference in an inbound message with the payload it points at. /// Runs before [`decode_payloads`], since a driver returns a payload as it was before offloading. -/// Without external storage to retrieve them, a reference is an error instead. +/// Without external storage to retrieve them, which includes every build without the +/// `experimental` feature, a reference is an error instead. /// /// On error the message is left partly processed and must not be used. #[doc(hidden)] @@ -158,9 +180,14 @@ pub async fn retrieve_external_payloads( msg: &mut M, data_converter: &DataConverter, ) -> Result<(), ExternalStorageError> { - // External storage cannot be configured yet, so there is never anything to retrieve with. - let _ = data_converter; - crate::external_storage_references::reject_references_in(msg).await + #[cfg(feature = "experimental")] + return crate::external_storage_runner::retrieve_in(data_converter.external_storage(), msg) + .await; + #[cfg(not(feature = "experimental"))] + { + let _ = data_converter; + crate::external_storage_references::reject_references_in(msg).await + } } // Manual impl for Payload - visits itself as a single payload diff --git a/crates/sdk-core/tests/integ_tests/external_storage_tests.rs b/crates/sdk-core/tests/integ_tests/external_storage_tests.rs new file mode 100644 index 000000000..292527338 --- /dev/null +++ b/crates/sdk-core/tests/integ_tests/external_storage_tests.rs @@ -0,0 +1,582 @@ +use crate::common::{ + CoreWfStarter, eventually, get_integ_connection, integ_namespace, rand_6_chars, +}; +use anyhow::anyhow; +use futures::{ + TryStreamExt, + future::{self, BoxFuture}, +}; +use std::{ + collections::HashMap, + sync::{Arc, Mutex}, + time::Duration, +}; +use temporalio_client::{ + ActivityIdentifier, Client, ClientOptions, WorkflowDescribeOptions, + WorkflowExecuteUpdateOptions, WorkflowListOptions, WorkflowQueryOptions, WorkflowSignalOptions, + WorkflowStartOptions, + errors::{WorkflowGetResultError, WorkflowStartError}, + schedules::{CreateScheduleOptions, ScheduleAction, ScheduleDescriptionAction, ScheduleSpec}, +}; +use temporalio_common::{ + MemoValues, + data_converters::{ + DataConverter, ExternalStorage, ExternalStorageError, StorageDriver, StorageDriverClaim, + StorageDriverError, StorageDriverRetrieveContext, StorageDriverStoreContext, + }, + protos::temporal::api::{common::v1::Payload, history::v1::history_event::Attributes}, +}; +use temporalio_macros::{activities, workflow, workflow_methods}; +use temporalio_sdk::{ + ActivityOptions, Runtime, SyncWorkflowContext, WorkerOptions, WorkflowContext, + WorkflowContextView, WorkflowResult, + activities::{ActivityContext, ActivityError}, +}; +use tokio::sync::mpsc; + +const REFERENCE_MESSAGE_TYPE: &[u8] = b"temporal.api.sdk.v1.ExternalStorageReference"; + +#[derive(Default)] +struct InMemoryDriver { + stored: Mutex>, +} + +impl StorageDriver for InMemoryDriver { + fn name(&self) -> &str { + "memory" + } + + fn driver_type(&self) -> &str { + "test.memory" + } + + fn store( + &self, + _context: &StorageDriverStoreContext, + payloads: Vec, + ) -> BoxFuture<'static, Result, StorageDriverError>> { + let mut stored = self.stored.lock().unwrap(); + let claims = payloads + .into_iter() + .map(|payload| { + let key = format!("k{}", stored.len()); + stored.insert(key.clone(), payload); + StorageDriverClaim::new(HashMap::from([("key".to_owned(), key)])) + }) + .collect(); + Box::pin(future::ready(Ok(claims))) + } + + fn retrieve( + &self, + _context: &StorageDriverRetrieveContext, + claims: Vec, + ) -> BoxFuture<'static, Result, StorageDriverError>> { + let stored = self.stored.lock().unwrap(); + let payloads = claims + .iter() + .map(|claim| { + stored + .get(&claim.claim_data["key"]) + .cloned() + .ok_or_else(|| { + StorageDriverError::from(anyhow!( + "nothing stored for {:?}", + claim.claim_data + )) + }) + }) + .collect(); + Box::pin(future::ready(payloads)) + } +} + +struct UnreachableDriver; + +impl StorageDriver for UnreachableDriver { + fn name(&self) -> &str { + "unreachable" + } + + fn driver_type(&self) -> &str { + "test.unreachable" + } + + fn store( + &self, + _context: &StorageDriverStoreContext, + _payloads: Vec, + ) -> BoxFuture<'static, Result, StorageDriverError>> { + Box::pin(future::ready(Err(anyhow!("bucket unreachable").into()))) + } + + fn retrieve( + &self, + _context: &StorageDriverRetrieveContext, + _claims: Vec, + ) -> BoxFuture<'static, Result, StorageDriverError>> { + Box::pin(future::ready(Err(anyhow!("bucket unreachable").into()))) + } +} + +struct LargePayloadActivities; +#[activities] +impl LargePayloadActivities { + #[activity] + async fn echo(_ctx: ActivityContext, input: String) -> Result { + Ok(input) + } +} + +#[workflow] +#[derive(Default)] +struct LargePayloadWorkflow; +#[workflow_methods] +impl LargePayloadWorkflow { + #[run] + async fn run(ctx: &mut WorkflowContext, input: String) -> WorkflowResult { + let output = ctx + .execute_activity( + LargePayloadActivities::echo, + input, + ActivityOptions::start_to_close_timeout(Duration::from_secs(5)), + ) + .await?; + Ok(output) + } +} + +#[workflow] +#[derive(Default)] +struct LargeUpdateWorkflow { + done: bool, +} +#[workflow_methods] +impl LargeUpdateWorkflow { + #[run] + async fn run(ctx: &mut WorkflowContext) -> WorkflowResult<()> { + ctx.wait_condition(|s| s.done).await?; + Ok(()) + } + + #[update] + fn echo(&mut self, _ctx: &mut SyncWorkflowContext, input: String) -> String { + self.done = true; + input + } +} + +#[workflow] +#[derive(Default)] +struct LargeSignalWorkflow { + value: Option, +} +#[workflow_methods] +impl LargeSignalWorkflow { + #[run] + async fn run(ctx: &mut WorkflowContext) -> WorkflowResult { + ctx.wait_condition(|s| s.value.is_some()).await?; + Ok(ctx.state(|s| s.value.clone().unwrap_or_default())) + } + + #[signal] + fn deliver(&mut self, _ctx: &mut SyncWorkflowContext, value: String) { + self.value = Some(value); + } + + #[query] + fn current(&self, _ctx: &WorkflowContextView) -> String { + self.value.clone().unwrap_or_default() + } +} + +struct AsyncCompletionActivities { + task_token: mpsc::Sender>, +} +#[activities] +impl AsyncCompletionActivities { + #[activity] + async fn complete_later( + self: Arc, + ctx: ActivityContext, + ) -> Result { + let _ = self.task_token.send(ctx.info().task_token.clone()).await; + Err(ActivityError::WillCompleteAsync) + } +} + +#[workflow] +#[derive(Default)] +struct AsyncCompletionWorkflow; +#[workflow_methods] +impl AsyncCompletionWorkflow { + #[run] + async fn run(ctx: &mut WorkflowContext) -> WorkflowResult { + let output = ctx + .execute_activity( + AsyncCompletionActivities::complete_later, + (), + ActivityOptions::start_to_close_timeout(Duration::from_secs(30)), + ) + .await?; + Ok(output) + } +} + +fn is_reference(payload: &Payload) -> bool { + payload.metadata.get("messageType").map(Vec::as_slice) == Some(REFERENCE_MESSAGE_TYPE) +} + +async fn client_with_storage(driver: Arc) -> Client { + let storage = ExternalStorage::new(driver) + .unwrap() + .with_payload_size_threshold(1024); + let client_opts = ClientOptions::new(integ_namespace()) + .data_converter(DataConverter::default().with_external_storage(storage)) + .build(); + Client::new(get_integ_connection(None).await, client_opts).unwrap() +} + +#[tokio::test] +async fn large_payloads_round_trip_through_external_storage() { + let driver = Arc::new(InMemoryDriver::default()); + let client = client_with_storage(driver.clone()).await; + + let mut starter = + CoreWfStarter::new_with_overrides(LargePayloadWorkflow::name(), None, Some(client.clone())); + starter + .sdk_config + .register_activities(LargePayloadActivities); + starter + .sdk_config + .register_workflow::() + .unwrap(); + let workflow_id = starter.get_task_queue().to_owned(); + let mut worker = starter.worker().await; + + let input = "x".repeat(10_000); + let handle = worker + .submit_workflow( + LargePayloadWorkflow::run, + input.clone(), + WorkflowStartOptions::new(starter.get_task_queue(), workflow_id.clone()).build(), + ) + .await + .unwrap(); + worker.run_until_done().await.unwrap(); + + assert_eq!(handle.get_result(Default::default()).await.unwrap(), input); + // The workflow's input, the activity's input and result, and the workflow's result. + assert_eq!(driver.stored.lock().unwrap().len(), 4); + + let history = starter.get_history().await; + let mut workflow_input = None; + let mut activity_input = None; + let mut workflow_result = None; + for event in history.events { + match event.attributes { + Some(Attributes::WorkflowExecutionStartedEventAttributes(attrs)) => { + workflow_input = attrs.input.map(|p| p.payloads[0].clone()); + } + Some(Attributes::ActivityTaskScheduledEventAttributes(attrs)) => { + activity_input = attrs.input.map(|p| p.payloads[0].clone()); + } + Some(Attributes::WorkflowExecutionCompletedEventAttributes(attrs)) => { + workflow_result = attrs.result.map(|p| p.payloads[0].clone()); + } + _ => {} + } + } + assert!(is_reference(&workflow_input.unwrap())); + assert!(is_reference(&activity_input.unwrap())); + assert!(is_reference(&workflow_result.unwrap())); + + let client_without_storage = Client::new( + get_integ_connection(None).await, + ClientOptions::new(integ_namespace()).build(), + ) + .unwrap(); + let err = client_without_storage + .get_workflow_handle::(workflow_id) + .get_result(Default::default()) + .await + .unwrap_err(); + assert!( + matches!( + err, + WorkflowGetResultError::ExternalStorage(ExternalStorageError::NotConfigured) + ), + "got {err:?}" + ); +} + +#[tokio::test] +async fn large_update_arguments_and_results_round_trip_through_external_storage() { + let driver = Arc::new(InMemoryDriver::default()); + let client = client_with_storage(driver.clone()).await; + let task_queue = format!("large_update_{}", rand_6_chars()); + let runtime = Runtime::from_current_tokio(Default::default()).unwrap(); + let worker_options = WorkerOptions::new(task_queue.clone()) + .register_workflow::() + .unwrap() + .build(); + let worker = temporalio_sdk::Worker::new(&runtime, client.clone(), worker_options).unwrap(); + + let input = "y".repeat(10_000); + worker + .run_until(async { + let handle = client + .start_workflow( + LargeUpdateWorkflow::run, + (), + WorkflowStartOptions::new(task_queue.clone(), task_queue.clone()).build(), + ) + .await + .unwrap(); + let output = handle + .execute_update( + LargeUpdateWorkflow::echo, + input.clone(), + WorkflowExecuteUpdateOptions::default(), + ) + .await + .unwrap(); + assert_eq!(output, input); + handle.get_result(Default::default()).await.unwrap(); + }) + .await + .unwrap(); + + // The update's argument, stored by the client, and its result, stored by the worker. + assert_eq!(driver.stored.lock().unwrap().len(), 2); +} + +#[tokio::test] +async fn a_failing_store_is_reported_as_an_external_storage_error() { + let storage = ExternalStorage::new(Arc::new(UnreachableDriver)) + .unwrap() + .with_payload_size_threshold(1024); + let client_opts = ClientOptions::new(integ_namespace()) + .data_converter(DataConverter::default().with_external_storage(storage)) + .build(); + let client = Client::new(get_integ_connection(None).await, client_opts).unwrap(); + let task_queue = format!("failing_store_{}", rand_6_chars()); + + let Err(err) = client + .start_workflow( + LargePayloadWorkflow::run, + "x".repeat(10_000), + WorkflowStartOptions::new(task_queue.clone(), task_queue).build(), + ) + .await + else { + panic!("starting the workflow should fail when its input cannot be stored"); + }; + + assert!( + matches!( + err, + WorkflowStartError::ExternalStorage(ExternalStorageError::Driver(_)) + ), + "got {err:?}" + ); +} + +#[tokio::test] +async fn signals_queries_and_memos_round_trip_through_external_storage() { + let driver = Arc::new(InMemoryDriver::default()); + let client = client_with_storage(driver.clone()).await; + let task_queue = format!("large_signal_{}", rand_6_chars()); + let runtime = Runtime::from_current_tokio(Default::default()).unwrap(); + let worker_options = WorkerOptions::new(task_queue.clone()) + .register_workflow::() + .unwrap() + .build(); + let worker = temporalio_sdk::Worker::new(&runtime, client.clone(), worker_options).unwrap(); + + let value = "s".repeat(10_000); + let memo_value = "m".repeat(10_000); + let mut memo = MemoValues::new(); + memo.insert("big", memo_value.clone()); + worker + .run_until(async { + let handle = client + .start_workflow( + LargeSignalWorkflow::run, + (), + WorkflowStartOptions::new(task_queue.clone(), task_queue.clone()) + .memo(memo) + .build(), + ) + .await + .unwrap(); + handle + .signal( + LargeSignalWorkflow::deliver, + value.clone(), + WorkflowSignalOptions::default(), + ) + .await + .unwrap(); + let queried = handle + .query( + LargeSignalWorkflow::current, + (), + WorkflowQueryOptions::default(), + ) + .await + .unwrap(); + assert_eq!(queried, value); + assert_eq!(handle.get_result(Default::default()).await.unwrap(), value); + }) + .await + .unwrap(); + + let handle = client.get_workflow_handle::(task_queue.clone()); + let description = handle + .describe(WorkflowDescribeOptions::default()) + .await + .unwrap(); + assert_eq!( + description.memo().get::("big").unwrap(), + Some(memo_value.clone()) + ); + + let listed = eventually( + || { + let client = client.clone(); + let query = format!("WorkflowId = '{task_queue}'"); + async move { + let executions: Vec<_> = client + .list_workflows(query, WorkflowListOptions::default()) + .try_collect() + .await + .map_err(|e| e.to_string())?; + executions + .into_iter() + .next() + .ok_or_else(|| "workflow not listed yet".to_owned()) + } + }, + Duration::from_secs(10), + ) + .await + .unwrap(); + assert_eq!( + listed.memo().get::("big").unwrap(), + Some(memo_value) + ); + + let events = handle + .fetch_history(Default::default()) + .into_events() + .await + .unwrap(); + let mut memo_payload = None; + let mut signal_input = None; + for event in events { + match event.attributes { + Some(Attributes::WorkflowExecutionStartedEventAttributes(attrs)) => { + memo_payload = attrs.memo.and_then(|m| m.fields.get("big").cloned()); + } + Some(Attributes::WorkflowExecutionSignaledEventAttributes(attrs)) => { + signal_input = attrs.input.map(|p| p.payloads[0].clone()); + } + _ => {} + } + } + assert!(is_reference(&memo_payload.unwrap())); + assert!(is_reference(&signal_input.unwrap())); +} + +#[tokio::test] +async fn schedule_arguments_round_trip_through_external_storage() { + let driver = Arc::new(InMemoryDriver::default()); + let client = client_with_storage(driver.clone()).await; + let schedule_id = format!("large_schedule_{}", rand_6_chars()); + let input = "a".repeat(10_000); + + let handle = client + .create_schedule( + &schedule_id, + CreateScheduleOptions::builder() + .action(ScheduleAction::start_workflow( + LargePayloadWorkflow::run, + input.clone(), + "unused-task-queue", + format!("wf-{schedule_id}"), + )) + .spec(ScheduleSpec::from_interval(Duration::from_secs(3600))) + .paused(true) + .build(), + ) + .await + .unwrap(); + assert_eq!(driver.stored.lock().unwrap().len(), 1); + + let description = handle.describe(Default::default()).await.unwrap(); + let ScheduleDescriptionAction::StartWorkflow(action) = description.action() else { + panic!("expected a start workflow action"); + }; + let stored: String = action + .args() + .await + .unwrap() + .expect("input should be present"); + assert_eq!(stored, input); + + handle.delete(Default::default()).await.unwrap(); +} + +#[tokio::test] +async fn async_activity_results_round_trip_through_external_storage() { + let driver = Arc::new(InMemoryDriver::default()); + let client = client_with_storage(driver.clone()).await; + let task_queue = format!("large_async_activity_{}", rand_6_chars()); + let (token_tx, mut token_rx) = mpsc::channel(1); + let runtime = Runtime::from_current_tokio(Default::default()).unwrap(); + let worker_options = WorkerOptions::new(task_queue.clone()) + .register_workflow::() + .unwrap() + .register_activities(AsyncCompletionActivities { + task_token: token_tx, + }) + .build(); + let worker = temporalio_sdk::Worker::new(&runtime, client.clone(), worker_options).unwrap(); + + let result = "r".repeat(10_000); + worker + .run_until(async { + let handle = client + .start_workflow( + AsyncCompletionWorkflow::run, + (), + WorkflowStartOptions::new(task_queue.clone(), task_queue.clone()).build(), + ) + .await + .unwrap(); + let token = token_rx.recv().await.unwrap(); + client + .get_async_activity_handle(ActivityIdentifier::TaskToken(token.into())) + .complete(Some(result.clone()), Default::default()) + .await + .unwrap(); + assert_eq!(handle.get_result(Default::default()).await.unwrap(), result); + }) + .await + .unwrap(); + + let events = client + .get_workflow_handle::(task_queue) + .fetch_history(Default::default()) + .into_events() + .await + .unwrap(); + let activity_result = events.into_iter().find_map(|event| match event.attributes { + Some(Attributes::ActivityTaskCompletedEventAttributes(attrs)) => { + attrs.result.map(|p| p.payloads[0].clone()) + } + _ => None, + }); + assert!(is_reference(&activity_result.unwrap())); +} diff --git a/crates/sdk-core/tests/main.rs b/crates/sdk-core/tests/main.rs index 007ed83d2..cee1ba611 100644 --- a/crates/sdk-core/tests/main.rs +++ b/crates/sdk-core/tests/main.rs @@ -25,6 +25,7 @@ mod integ_tests { mod client_tests; mod data_converter_tests; mod ephemeral_server_tests; + mod external_storage_tests; mod heartbeat_tests; mod metrics_tests; mod pagination_tests; diff --git a/crates/sdk/src/lib.rs b/crates/sdk/src/lib.rs index 88f9205ef..0c494968b 100644 --- a/crates/sdk/src/lib.rs +++ b/crates/sdk/src/lib.rs @@ -151,7 +151,9 @@ use temporalio_common::{ ActivitySerializationContext, DataConverter, SerializationContext, SerializationContextData, WorkflowSerializationContext, }, - payload_visitor::{decode_payloads, encode_payloads, retrieve_external_payloads}, + payload_visitor::{ + decode_payloads, encode_payloads, retrieve_external_payloads, store_external_payloads, + }, protos::{ TaskToken, coresdk::{ @@ -865,13 +867,18 @@ async fn encode_workflow_completion( data_converter: &DataConverter, ) { let run_id = completion.run_id.clone(); - if let Err(err) = encode_payloads( - completion, - data_converter.codec(), - &SerializationContextData::Workflow(WorkflowSerializationContext::new()), - ) - .await - { + let encoded = async { + encode_payloads( + completion, + data_converter.codec(), + &SerializationContextData::Workflow(WorkflowSerializationContext::new()), + ) + .await?; + store_external_payloads(completion, data_converter).await?; + Ok::<_, anyhow::Error>(()) + } + .await; + if let Err(err) = encoded { error!(run_id, error = %err, "Failed encoding workflow activation completion"); *completion = WorkflowActivationCompletion::fail( run_id, @@ -888,13 +895,18 @@ async fn encode_activity_completion( completion: &mut ActivityTaskCompletion, data_converter: &DataConverter, ) { - if let Err(err) = encode_payloads( - completion, - data_converter.codec(), - &SerializationContextData::Activity(ActivitySerializationContext::new()), - ) - .await - { + let encoded = async { + encode_payloads( + completion, + data_converter.codec(), + &SerializationContextData::Activity(ActivitySerializationContext::new()), + ) + .await?; + store_external_payloads(completion, data_converter).await?; + Ok::<_, anyhow::Error>(()) + } + .await; + if let Err(err) = encoded { error!(error = %err, "Failed encoding activity task completion"); completion.result = Some(ActivityExecutionResult::fail(Failure::application_failure( format!("Failed encoding activity completion: {err}"),