feat: offload large payloads via external storage

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.

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"]

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<ActivityT::Output, ActivityResultError> {
         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<ActivityExecutionDescription<ActivityT>, 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(),

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<CT: WorkflowService + NamespacedClient + Clone> AsyncActivityHandle<CT> {
                                 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<CT: WorkflowService + NamespacedClient + Clone> AsyncActivityHandle<CT> {
                                 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<CT: WorkflowService + NamespacedClient + Clone> AsyncActivityHandle<CT> {
                                 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<CT: WorkflowService + NamespacedClient + Clone> AsyncActivityHandle<CT> {
                                 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<CT: WorkflowService + NamespacedClient + Clone> AsyncActivityHandle<CT> {
                                 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<CT: WorkflowService + NamespacedClient + Clone> AsyncActivityHandle<CT> {
                                 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<CT: WorkflowService + NamespacedClient + Clone> AsyncActivityHandle<CT> {
                                 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<CT: WorkflowService + NamespacedClient + Clone> AsyncActivityHandle<CT> {
                                 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(

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<WorkflowInteractionError> 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),