Skip to content

Commit e8f8747

Browse files
raphael-goetzraphael-goetz
committed
fix: normalized execution result
1 parent 7710540 commit e8f8747

1 file changed

Lines changed: 241 additions & 2 deletions

File tree

‎crates/taurus/src/client/runtime_execution.rs‎

Lines changed: 241 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,10 @@ use code0_flow::flow_service::{
44
use tonic::{Extensions, Request, transport::Channel};
55
use tucana::{
66
aquila::{ExecutionRequest, execution_service_client::ExecutionServiceClient},
7-
shared::ExecutionResult,
7+
shared::{
8+
ExecutionResult, NodeExecutionResult, Value, execution_result, node_execution_result,
9+
value::Kind,
10+
},
811
};
912

1013
pub struct TaurusRuntimeExecutionService {
@@ -23,14 +26,18 @@ impl TaurusRuntimeExecutionService {
2326
}
2427
}
2528

26-
pub async fn update_runtime_execution(&mut self, runtime_execution: ExecutionResult) {
29+
pub async fn update_runtime_execution(&mut self, mut runtime_execution: ExecutionResult) {
2730
log::info!(
2831
"Transmitting execution result to Aquila (execution_id={}, flow_id={}, node_results={})",
2932
runtime_execution.execution_identifier.as_str(),
3033
runtime_execution.flow_id,
3134
runtime_execution.node_execution_results.len()
3235
);
3336

37+
normalize_execution_result(&mut runtime_execution);
38+
39+
log::debug!("{:#?}", runtime_execution);
40+
3441
let request = Request::from_parts(
3542
get_authorization_metadata(&self.aquila_token),
3643
Extensions::new(),
@@ -52,3 +59,235 @@ impl TaurusRuntimeExecutionService {
5259
}
5360
}
5461
}
62+
63+
fn normalize_execution_result(result: &mut ExecutionResult) {
64+
match &mut result.input {
65+
Some(input) => normalize_value(input),
66+
None => {
67+
result.input = Some(null_value());
68+
}
69+
}
70+
71+
for node_result in &mut result.node_execution_results {
72+
normalize_node_execution_result(node_result);
73+
}
74+
75+
match &mut result.result {
76+
Some(execution_result::Result::Success(value)) => normalize_value(value),
77+
Some(execution_result::Result::Error(error)) => {
78+
if let Some(details) = &mut error.details {
79+
for value in details.fields.values_mut() {
80+
normalize_value(value);
81+
}
82+
}
83+
}
84+
None => {
85+
result.result = Some(execution_result::Result::Success(null_value()));
86+
}
87+
}
88+
}
89+
90+
fn normalize_node_execution_result(result: &mut NodeExecutionResult) {
91+
for parameter_result in &mut result.parameter_results {
92+
match &mut parameter_result.value {
93+
Some(value) => normalize_value(value),
94+
None => {
95+
parameter_result.value = Some(null_value());
96+
}
97+
}
98+
}
99+
100+
match &mut result.result {
101+
Some(node_execution_result::Result::Success(value)) => normalize_value(value),
102+
Some(node_execution_result::Result::Error(error)) => {
103+
if let Some(details) = &mut error.details {
104+
for value in details.fields.values_mut() {
105+
normalize_value(value);
106+
}
107+
}
108+
}
109+
None => {
110+
result.result = Some(node_execution_result::Result::Success(null_value()));
111+
}
112+
}
113+
}
114+
115+
fn normalize_value(value: &mut Value) {
116+
match &mut value.kind {
117+
Some(Kind::StructValue(struct_value)) => {
118+
for field in struct_value.fields.values_mut() {
119+
normalize_value(field);
120+
}
121+
}
122+
Some(Kind::ListValue(list_value)) => {
123+
for item in &mut list_value.values {
124+
normalize_value(item);
125+
}
126+
}
127+
Some(Kind::NumberValue(number)) if number.number.is_none() => {
128+
value.kind = Some(Kind::NullValue(0));
129+
}
130+
Some(_) => {}
131+
None => {
132+
value.kind = Some(Kind::NullValue(0));
133+
}
134+
}
135+
}
136+
137+
fn null_value() -> Value {
138+
Value {
139+
kind: Some(Kind::NullValue(0)),
140+
}
141+
}
142+
143+
#[cfg(test)]
144+
mod tests {
145+
use super::*;
146+
use tucana::shared::{
147+
Error, ListValue, NodeParameterNodeExecutionResult, NumberValue, Struct,
148+
node_execution_result,
149+
};
150+
151+
#[test]
152+
fn normalize_execution_result_fills_missing_results_with_null_success() {
153+
let mut result = ExecutionResult {
154+
execution_identifier: "execution-id".to_string(),
155+
flow_id: 1,
156+
started_at: 1,
157+
finished_at: 2,
158+
input: None,
159+
node_execution_results: vec![NodeExecutionResult {
160+
started_at: 1,
161+
finished_at: 2,
162+
parameter_results: vec![NodeParameterNodeExecutionResult { value: None }],
163+
id: Some(node_execution_result::Id::NodeId(10)),
164+
result: None,
165+
}],
166+
result: None,
167+
};
168+
169+
normalize_execution_result(&mut result);
170+
171+
assert!(matches!(
172+
result.input.as_ref(),
173+
Some(Value {
174+
kind: Some(Kind::NullValue(_))
175+
})
176+
));
177+
assert!(matches!(
178+
result.result.as_ref(),
179+
Some(execution_result::Result::Success(Value {
180+
kind: Some(Kind::NullValue(_))
181+
}))
182+
));
183+
assert!(matches!(
184+
result.node_execution_results[0].result.as_ref(),
185+
Some(node_execution_result::Result::Success(Value {
186+
kind: Some(Kind::NullValue(_))
187+
}))
188+
));
189+
assert!(matches!(
190+
result.node_execution_results[0].parameter_results[0]
191+
.value
192+
.as_ref(),
193+
Some(Value {
194+
kind: Some(Kind::NullValue(_))
195+
})
196+
));
197+
}
198+
199+
#[test]
200+
fn normalize_execution_result_recurses_into_child_values() {
201+
let mut fields = std::collections::HashMap::new();
202+
fields.insert(
203+
"list".to_string(),
204+
Value {
205+
kind: Some(Kind::ListValue(ListValue {
206+
values: vec![
207+
Value { kind: None },
208+
Value {
209+
kind: Some(Kind::NumberValue(NumberValue { number: None })),
210+
},
211+
],
212+
})),
213+
},
214+
);
215+
let mut result = ExecutionResult {
216+
execution_identifier: "execution-id".to_string(),
217+
flow_id: 1,
218+
started_at: 1,
219+
finished_at: 2,
220+
input: Some(Value {
221+
kind: Some(Kind::StructValue(Struct { fields })),
222+
}),
223+
node_execution_results: vec![NodeExecutionResult {
224+
started_at: 1,
225+
finished_at: 2,
226+
parameter_results: vec![NodeParameterNodeExecutionResult {
227+
value: Some(Value { kind: None }),
228+
}],
229+
id: Some(node_execution_result::Id::NodeId(10)),
230+
result: Some(node_execution_result::Result::Error(Error {
231+
details: Some(Struct {
232+
fields: std::collections::HashMap::from([(
233+
"empty".to_string(),
234+
Value { kind: None },
235+
)]),
236+
}),
237+
..Default::default()
238+
})),
239+
}],
240+
result: Some(execution_result::Result::Success(Value { kind: None })),
241+
};
242+
243+
normalize_execution_result(&mut result);
244+
245+
assert!(matches!(
246+
result.result.as_ref(),
247+
Some(execution_result::Result::Success(Value {
248+
kind: Some(Kind::NullValue(_))
249+
}))
250+
));
251+
assert!(matches!(
252+
result.node_execution_results[0].parameter_results[0]
253+
.value
254+
.as_ref(),
255+
Some(Value {
256+
kind: Some(Kind::NullValue(_))
257+
})
258+
));
259+
let Some(Value {
260+
kind: Some(Kind::StructValue(input)),
261+
}) = result.input.as_ref()
262+
else {
263+
panic!("expected normalized struct input");
264+
};
265+
let Some(Value {
266+
kind: Some(Kind::ListValue(list)),
267+
}) = input.fields.get("list")
268+
else {
269+
panic!("expected normalized list field");
270+
};
271+
assert!(matches!(
272+
list.values[0].kind.as_ref(),
273+
Some(Kind::NullValue(_))
274+
));
275+
assert!(matches!(
276+
list.values[1].kind.as_ref(),
277+
Some(Kind::NullValue(_))
278+
));
279+
let Some(node_execution_result::Result::Error(error)) =
280+
result.node_execution_results[0].result.as_ref()
281+
else {
282+
panic!("expected node error result");
283+
};
284+
assert!(matches!(
285+
error
286+
.details
287+
.as_ref()
288+
.and_then(|details| details.fields.get("empty"))
289+
.and_then(|value| value.kind.as_ref()),
290+
Some(Kind::NullValue(_))
291+
));
292+
}
293+
}

0 commit comments

Comments
 (0)