Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
44 commits
Select commit Hold shift + click to select a range
9305879
initial plan
chris-olszewski Apr 12, 2026
9a31920
Add ApplicationFailure and reshape ActivityError
chris-olszewski Apr 12, 2026
59f2ec3
Move shared errors and split data_converters
chris-olszewski Apr 12, 2026
347bf9c
Implement erased failure classification
chris-olszewski Apr 12, 2026
0f22ec3
Route workflow/activity failures through converter and update nexus a…
chris-olszewski Apr 12, 2026
1c15e1e
fix: do not nest application failures
chris-olszewski Apr 13, 2026
4c3f6b2
doc: failure converter plan
chris-olszewski Apr 13, 2026
6ae267b
add activity error hint
chris-olszewski Apr 16, 2026
078ec66
add child workflow decoding
chris-olszewski Apr 16, 2026
545c4cd
child workflow signal decode
chris-olszewski Apr 16, 2026
631eb6f
use outgoing error type
chris-olszewski Apr 16, 2026
7da9e94
differentiate workflow failures
chris-olszewski Apr 17, 2026
2a0e692
enumerate all current failure info options
chris-olszewski Apr 17, 2026
985a6f9
hook up query/update handlers
chris-olszewski Apr 17, 2026
03890ec
richer failure error shapes
chris-olszewski Apr 17, 2026
7dc48a9
activity errors tightening
chris-olszewski Apr 20, 2026
ad0b641
surface cancellation failure info
chris-olszewski Apr 20, 2026
f0923ac
enhance child workflow execution error
chris-olszewski Apr 20, 2026
52cf033
child signal error type
chris-olszewski Apr 20, 2026
1e5049c
update docs
chris-olszewski Apr 20, 2026
b16c9c2
flesh out activity failure
chris-olszewski Apr 20, 2026
f6a5d3e
prep for child wfl split
chris-olszewski Apr 21, 2026
d97741e
split child start/execution error type
chris-olszewski Apr 21, 2026
f2dd70b
refactor tests
chris-olszewski Apr 21, 2026
3a8ccbe
failure converter refactor
chris-olszewski Apr 21, 2026
c31843d
add module doc comment
chris-olszewski Apr 21, 2026
88a7b7b
remove arch docs
chris-olszewski Apr 21, 2026
05ad2aa
collapse activity failure branches
chris-olszewski Apr 21, 2026
2492475
pr feedback
chris-olszewski Apr 22, 2026
ddad318
switch from mod.rs to named module file
chris-olszewski Apr 22, 2026
7d2fe28
serialize details to payload
chris-olszewski Apr 23, 2026
743eadd
chore: add ergonomic constructors for cancellation details
chris-olszewski Apr 24, 2026
3bc0556
remove arch doc
chris-olszewski Apr 24, 2026
c4ae99d
fix cause drops on conversion
chris-olszewski Apr 24, 2026
9fe1c84
remove redundant test
chris-olszewski Apr 24, 2026
f995f29
use payload converter from data converter
chris-olszewski Apr 25, 2026
a69437b
simplify workflow_message_to_failure
chris-olszewski Apr 25, 2026
f67456c
remove useless LA unit test
chris-olszewski Apr 25, 2026
4e6f381
include basic callout of failure conversion in readme
chris-olszewski Apr 25, 2026
19429b4
fix lints
chris-olszewski Apr 25, 2026
6aa1bf8
fix formatting
chris-olszewski Apr 25, 2026
801bcf3
add identitiy to cancellation
chris-olszewski Apr 28, 2026
db7c2f1
fix merge conflict
chris-olszewski Apr 29, 2026
0bcfb67
special case ApplicationFailure during ActivityError construciton
chris-olszewski Apr 29, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions crates/common/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -107,5 +107,6 @@ workspace = true

[dev-dependencies]
futures-util = { version = "0.3", default-features = false }
rstest = "0.26"
tempfile = "3.21"
tokio = { version = "1.47", features = ["macros", "rt"] }
172 changes: 133 additions & 39 deletions crates/common/src/data_converters.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,14 @@
//! Contains traits for and default implementations of data converters, codecs, and other
//! serialization related functionality.

use crate::protos::temporal::api::{common::v1::Payload, failure::v1::Failure};
mod failure_converter;
Comment thread
chris-olszewski marked this conversation as resolved.

pub use failure_converter::{
ActivityExecutionDecodeHint, ChildWorkflowExecutionDecodeHint, ChildWorkflowSignalDecodeHint,
ChildWorkflowStartDecodeHint, DefaultFailureConverter, FailureConverter, FailureDecodeHint,
};

use crate::protos::temporal::api::common::v1::Payload;
use futures::{FutureExt, future::BoxFuture};
use std::{collections::HashMap, sync::Arc};

Expand All @@ -22,6 +29,7 @@ impl std::fmt::Debug for DataConverter {
.finish_non_exhaustive()
}
}

impl DataConverter {
/// Create a new DataConverter with the given payload converter, failure converter, and codec.
pub fn new(
Expand Down Expand Up @@ -105,6 +113,34 @@ impl DataConverter {
&self.payload_converter
}

/// Returns the failure converter component of this data converter.
pub fn failure_converter(&self) -> &(dyn FailureConverter + Send + Sync) {
self.failure_converter.as_ref()
}

/// Decode a Temporal failure into a caller-facing Rust error surface.
pub fn to_error<H: FailureDecodeHint>(
&self,
context: &SerializationContextData,
failure: crate::protos::temporal::api::failure::v1::Failure,
hint: H,
) -> Result<H::Output, PayloadConversionError> {
let normalized =
self.failure_converter
.to_error(failure, &self.payload_converter, context)?;
Ok(hint.adapt(normalized))
}

/// Encode a typed Rust error surface into a Temporal failure.
pub fn to_failure(
&self,
context: &SerializationContextData,
error: crate::error::OutgoingError,
) -> crate::protos::temporal::api::failure::v1::Failure {
self.failure_converter
.to_failure(error, &self.payload_converter, context)
}

/// Returns the codec component of this data converter.
pub fn codec(&self) -> &(dyn PayloadCodec + Send + Sync) {
self.codec.as_ref()
Expand Down Expand Up @@ -196,26 +232,6 @@ impl std::error::Error for PayloadConversionError {
}
}

/// Converts between Rust errors and Temporal [`Failure`] protobufs.
pub trait FailureConverter {
/// Convert an error into a Temporal failure protobuf.
fn to_failure(
&self,
error: Box<dyn std::error::Error>,
payload_converter: &PayloadConverter,
context: &SerializationContextData,
) -> Result<Failure, PayloadConversionError>;

/// Convert a Temporal failure protobuf back into a Rust error.
fn to_error(
&self,
failure: Failure,
payload_converter: &PayloadConverter,
context: &SerializationContextData,
) -> Result<Box<dyn std::error::Error>, PayloadConversionError>;
}
/// Default (currently unimplemented) failure converter.
pub struct DefaultFailureConverter;
/// Encodes and decodes payloads, enabling encryption or compression.
pub trait PayloadCodec {
/// Encode payloads before they are sent to the server.
Expand Down Expand Up @@ -307,6 +323,53 @@ pub trait TemporalDeserializable: Sized {
}
}

/// A codec-decoded set of payloads that can be deserialized later with to a user provided type.
#[derive(Clone, Debug)]
pub struct DecodablePayloads {
payloads: Vec<Payload>,
payload_converter: PayloadConverter,
context: SerializationContextData,
}

impl DecodablePayloads {
/// Create a new decodable payload set from raw payloads and the converter context needed to
/// deserialize them later.
pub fn new(
payloads: Vec<Payload>,
payload_converter: PayloadConverter,
context: SerializationContextData,
) -> Self {
Self {
payloads,
payload_converter,
context,
}
}

/// Deserialize these payloads into a typed value using the stored payload converter.
pub fn deserialize<T: TemporalDeserializable + 'static>(
&self,
) -> Result<T, PayloadConversionError> {
self.payload_converter.from_payloads(
&SerializationContext {
data: &self.context,
converter: &self.payload_converter,
},
self.payloads.clone(),
)
}

/// Returns the underlying payloads.
pub fn raw(&self) -> &[Payload] {
&self.payloads
}

/// Consume this value and return the underlying payloads as a [`RawValue`].
pub fn into_raw(self) -> RawValue {
RawValue::new(self.payloads)
}
}

/// An unconverted set of payloads, used when the caller wants to defer deserialization.
#[derive(Clone, Debug, Default)]
pub struct RawValue {
Expand Down Expand Up @@ -672,24 +735,6 @@ impl Default for DataConverter {
)
}
}
impl FailureConverter for DefaultFailureConverter {
fn to_failure(
&self,
_: Box<dyn std::error::Error>,
_: &PayloadConverter,
_: &SerializationContextData,
) -> Result<Failure, PayloadConversionError> {
todo!()
}
fn to_error(
&self,
_: Failure,
_: &PayloadConverter,
_: &SerializationContextData,
) -> Result<Box<dyn std::error::Error>, PayloadConversionError> {
todo!()
}
}
impl PayloadCodec for DefaultPayloadCodec {
fn encode(
&self,
Expand Down Expand Up @@ -866,4 +911,53 @@ mod tests {
let args: MultiArgs2<String, i32> = ("hello".to_string(), 42i32).into();
assert_eq!(args, MultiArgs2("hello".to_string(), 42));
}

fn decodable_from_value<T: TemporalSerializable + 'static>(value: &T) -> DecodablePayloads {
let converter = PayloadConverter::default();
let payloads = converter
.to_payloads(
&SerializationContext {
data: &SerializationContextData::Workflow,
converter: &converter,
},
value,
)
.unwrap();
DecodablePayloads::new(payloads, converter, SerializationContextData::Workflow)
}
#[test]
fn decodable_payloads_roundtrip_string() {
let payloads = decodable_from_value(&"hello".to_string());

let result: String = payloads.deserialize().unwrap();

assert_eq!(result, "hello");
}

#[test]
fn decodable_payloads_roundtrip_option_string() {
let payloads = decodable_from_value(&Some("hello".to_string()));

let result: Option<String> = payloads.deserialize().unwrap();

assert_eq!(result, Some("hello".to_string()));
}

#[test]
fn decodable_payloads_roundtrip_unit() {
let payloads = decodable_from_value(&());

let result: () = payloads.deserialize().unwrap();

assert_eq!(result, ());
}

#[test]
fn decodable_payloads_roundtrip_vec_string() {
let payloads = decodable_from_value(&vec!["hello".to_string(), "world".to_string()]);

let result: Vec<String> = payloads.deserialize().unwrap();

assert_eq!(result, vec!["hello".to_string(), "world".to_string()]);
}
}
Loading
Loading