aion_proto/
workflow.rs

1//! Workflow-management serde/prost wire types.
2
3use crate::convert::{ProtoPayload, ProtoRunId, ProtoWorkflowId, WireEnvelope};
4use crate::error::ProtoWireError;
5
6/// Proto representation of `StartWorkflowRequest`.
7#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
8pub struct ProtoStartWorkflowRequest {
9    /// Namespace that scopes the operation.
10    #[prost(string, tag = "1")]
11    pub namespace: String,
12    /// Workflow type name registered with the engine.
13    #[prost(string, tag = "2")]
14    pub workflow_type: String,
15    /// Workflow start input payload.
16    #[prost(message, optional, tag = "3")]
17    pub input: Option<ProtoPayload>,
18    /// R-4 steered-start routing key. When set, the start is steered to
19    /// `shard_for(routing_key)`'s owner (forwarded there when this node is not the
20    /// owner). `None`/empty keeps the unsteered R-1 remint behaviour.
21    #[prost(string, optional, tag = "4")]
22    pub routing_key: Option<String>,
23    /// Optional task queue this workflow defaults its activities to (the
24    /// namespace × `task_queue` targeting story). When set, the server records it
25    /// durably on the start so it survives replay/failover. `None`/empty keeps
26    /// the namespace's default queue.
27    #[prost(string, optional, tag = "5")]
28    pub task_queue: Option<String>,
29    /// Optional operator-facing display name for the workflow this start
30    /// creates (#211). A LABEL over the UUID identity, never an address:
31    /// nothing resolves a workflow by name. When set, the server records it
32    /// durably (as the `aion.display_name` search attribute) in the same atomic
33    /// append as the start. `None` starts it unnamed; a present but blank
34    /// (empty or whitespace-only) value is REFUSED with `invalid_input` rather
35    /// than reinterpreted as unnamed, so pass `None` to mean "no name" — never
36    /// `Some("")`. Unlike `routing_key` and `task_queue`, empty here is not
37    /// "not selected".
38    ///
39    /// The recorded attribute carries no run id, so it reads back per WORKFLOW
40    /// (folded over the whole history, last write wins) rather than per run.
41    #[prost(string, optional, tag = "6")]
42    pub display_name: Option<String>,
43}
44
45/// Proto representation of `StartWorkflowResponse`.
46#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
47pub struct ProtoStartWorkflowResponse {
48    /// Assigned workflow identifier.
49    #[prost(message, optional, tag = "1")]
50    pub workflow_id: Option<ProtoWorkflowId>,
51    /// Assigned concrete run identifier.
52    #[prost(message, optional, tag = "2")]
53    pub run_id: Option<ProtoRunId>,
54}
55
56/// Proto representation of `SignalRequest`.
57#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
58pub struct ProtoSignalRequest {
59    /// Namespace that scopes the operation.
60    #[prost(string, tag = "1")]
61    pub namespace: String,
62    /// Target workflow identifier.
63    #[prost(message, optional, tag = "2")]
64    pub workflow_id: Option<ProtoWorkflowId>,
65    /// Target run identifier.
66    #[prost(message, optional, tag = "3")]
67    pub run_id: Option<ProtoRunId>,
68    /// Signal name registered by workflow code.
69    #[prost(string, tag = "4")]
70    pub signal_name: String,
71    /// Signal payload.
72    #[prost(message, optional, tag = "5")]
73    pub payload: Option<ProtoPayload>,
74}
75
76/// Proto representation of `SignalResponse`.
77#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
78pub struct ProtoSignalResponse {}
79
80/// Proto representation of `QueryRequest`.
81#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
82pub struct ProtoQueryRequest {
83    /// Namespace that scopes the operation.
84    #[prost(string, tag = "1")]
85    pub namespace: String,
86    /// Target workflow identifier.
87    #[prost(message, optional, tag = "2")]
88    pub workflow_id: Option<ProtoWorkflowId>,
89    /// Target run identifier.
90    #[prost(message, optional, tag = "3")]
91    pub run_id: Option<ProtoRunId>,
92    /// Query name registered by workflow code.
93    #[prost(string, tag = "4")]
94    pub query_name: String,
95    /// Caller-supplied arguments handed to the workflow's query handler.
96    ///
97    /// Absent means "no arguments": the server materializes the JSON `null`
98    /// document so a handler always receives one well-formed input. Arguments
99    /// are read-only handler inputs and are never recorded in history.
100    #[prost(message, optional, tag = "5")]
101    pub arguments: Option<ProtoPayload>,
102}
103
104/// Proto representation of `QueryResponse`.
105#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
106pub struct ProtoQueryResponse {
107    /// Query result or typed wire error.
108    #[prost(oneof = "proto_query_response::Outcome", tags = "1, 2")]
109    pub outcome: Option<proto_query_response::Outcome>,
110}
111
112/// Types nested under [`ProtoQueryResponse`].
113pub mod proto_query_response {
114    /// Proto oneof for successful query payloads and typed failures.
115    #[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Oneof)]
116    pub enum Outcome {
117        /// Query result payload.
118        #[prost(message, tag = "1")]
119        Result(super::ProtoPayload),
120        /// Typed query error.
121        #[prost(message, tag = "2")]
122        Error(super::ProtoWireError),
123    }
124}
125
126/// Proto representation of `CancelRequest`.
127#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
128pub struct ProtoCancelRequest {
129    /// Namespace that scopes the operation.
130    #[prost(string, tag = "1")]
131    pub namespace: String,
132    /// Target workflow identifier.
133    #[prost(message, optional, tag = "2")]
134    pub workflow_id: Option<ProtoWorkflowId>,
135    /// Target run identifier.
136    #[prost(message, optional, tag = "3")]
137    pub run_id: Option<ProtoRunId>,
138    /// Human-readable cancellation reason.
139    #[prost(string, tag = "4")]
140    pub reason: String,
141}
142
143/// Proto representation of `CancelResponse`.
144#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
145pub struct ProtoCancelResponse {}
146
147/// Proto representation of `ReopenRequest`.
148///
149/// Mirrors [`ProtoCancelRequest`] without a `reason`: the reopen carries only a
150/// target. An absent `run_id` means the latest run.
151#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
152pub struct ProtoReopenRequest {
153    /// Namespace that scopes the operation.
154    #[prost(string, tag = "1")]
155    pub namespace: String,
156    /// Target workflow identifier.
157    #[prost(message, optional, tag = "2")]
158    pub workflow_id: Option<ProtoWorkflowId>,
159    /// Target run identifier (absent means the latest run).
160    #[prost(message, optional, tag = "3")]
161    pub run_id: Option<ProtoRunId>,
162}
163
164/// Proto representation of `ReopenResponse`.
165///
166/// Unlike [`ProtoCancelResponse`] (an empty ack) this returns the reopened run
167/// id and its projected status (Running) so the caller learns the run is live
168/// again.
169#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
170pub struct ProtoReopenResponse {
171    /// The reopened concrete run identifier.
172    #[prost(message, optional, tag = "1")]
173    pub run_id: Option<ProtoRunId>,
174    /// The projected workflow status after the reopen (Running).
175    #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
176    pub status: i32,
177}
178
179/// Proto representation of `PauseRequest` (#204).
180///
181/// Mirrors [`ProtoCancelRequest`]: a target plus an optional reason. An absent
182/// `run_id` means the latest run.
183#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
184pub struct ProtoPauseRequest {
185    /// Namespace that scopes the operation.
186    #[prost(string, tag = "1")]
187    pub namespace: String,
188    /// Target workflow identifier.
189    #[prost(message, optional, tag = "2")]
190    pub workflow_id: Option<ProtoWorkflowId>,
191    /// Target run identifier (absent means the latest run).
192    #[prost(message, optional, tag = "3")]
193    pub run_id: Option<ProtoRunId>,
194    /// Optional operator-supplied pause reason.
195    #[prost(string, tag = "4")]
196    pub reason: String,
197}
198
199/// Proto representation of `PauseResponse` (#204): the paused run and its status.
200#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
201pub struct ProtoPauseResponse {
202    /// The paused concrete run identifier.
203    #[prost(message, optional, tag = "1")]
204    pub run_id: Option<ProtoRunId>,
205    /// The projected workflow status after the pause (Paused).
206    #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
207    pub status: i32,
208}
209
210/// Proto representation of `ResumeRequest` (#204).
211///
212/// Mirrors [`ProtoReopenRequest`]: only a target. An absent `run_id` means the
213/// latest run.
214#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
215pub struct ProtoResumeRequest {
216    /// Namespace that scopes the operation.
217    #[prost(string, tag = "1")]
218    pub namespace: String,
219    /// Target workflow identifier.
220    #[prost(message, optional, tag = "2")]
221    pub workflow_id: Option<ProtoWorkflowId>,
222    /// Target run identifier (absent means the latest run).
223    #[prost(message, optional, tag = "3")]
224    pub run_id: Option<ProtoRunId>,
225}
226
227/// Proto representation of `ResumeResponse` (#204): the resumed run and status.
228#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
229pub struct ProtoResumeResponse {
230    /// The resumed concrete run identifier.
231    #[prost(message, optional, tag = "1")]
232    pub run_id: Option<ProtoRunId>,
233    /// The projected workflow status after the resume (Running).
234    #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
235    pub status: i32,
236}
237
238/// Proto representation of `RenameRequest` (#211).
239///
240/// Mirrors [`ProtoPauseRequest`]'s shape: a target plus the operator's payload
241/// (here the new display name). An absent `run_id` means the latest run. The
242/// name is a LABEL over the UUID identity, never an address — this request
243/// SETS a name on an id-addressed run; nothing resolves a workflow by name.
244///
245/// Both facts hold at once, and they are easy to confuse: you ADDRESS a run
246/// (workflow id plus run id, so the server can refuse a rename it cannot append
247/// safely), but what gets recorded is a WORKFLOW-level attribute with no run id
248/// in it. The name therefore reads back for every run of that workflow, which
249/// is exactly why the engine refuses to rename a continue-as-new predecessor:
250/// its name would land on the successor that now owns the history head.
251#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
252pub struct ProtoRenameRequest {
253    /// Namespace that scopes the operation.
254    #[prost(string, tag = "1")]
255    pub namespace: String,
256    /// Target workflow identifier.
257    #[prost(message, optional, tag = "2")]
258    pub workflow_id: Option<ProtoWorkflowId>,
259    /// Target run identifier (absent means the latest run).
260    #[prost(message, optional, tag = "3")]
261    pub run_id: Option<ProtoRunId>,
262    /// The new display name. Trimmed by the server; must be non-empty after
263    /// trimming.
264    #[prost(string, tag = "4")]
265    pub display_name: String,
266}
267
268/// Proto representation of `RenameResponse` (#211): the renamed run and the
269/// display name as recorded (trimmed).
270#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
271pub struct ProtoRenameResponse {
272    /// The renamed concrete run identifier.
273    #[prost(message, optional, tag = "1")]
274    pub run_id: Option<ProtoRunId>,
275    /// The display name as recorded (trimmed).
276    #[prost(string, tag = "2")]
277    pub display_name: String,
278}
279
280/// Proto representation of `ListWorkflowsRequest`.
281#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
282pub struct ProtoListWorkflowsRequest {
283    /// Namespace that scopes the operation.
284    #[prost(string, tag = "1")]
285    pub namespace: String,
286    /// Serde-encoded `aion_store::visibility::ListWorkflowsFilter` envelope.
287    #[prost(message, optional, tag = "2")]
288    pub filter: Option<WireEnvelope>,
289}
290
291/// Proto representation of `ListWorkflowsResponse`.
292#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
293pub struct ProtoListWorkflowsResponse {
294    /// Serde-encoded `aion_store::visibility::WorkflowSummary` envelopes.
295    #[prost(message, repeated, tag = "1")]
296    pub summaries: Vec<WireEnvelope>,
297}
298
299/// Proto representation of `CountWorkflowsRequest`.
300#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
301pub struct ProtoCountWorkflowsRequest {
302    /// Namespace that scopes the operation.
303    #[prost(string, tag = "1")]
304    pub namespace: String,
305    /// Serde-encoded `aion_store::visibility::ListWorkflowsFilter` envelope.
306    #[prost(message, optional, tag = "2")]
307    pub filter: Option<WireEnvelope>,
308}
309
310/// Proto representation of `CountWorkflowsResponse`.
311#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
312pub struct ProtoCountWorkflowsResponse {
313    /// Number of visibility summaries matching the filter.
314    #[prost(uint64, tag = "1")]
315    pub count: u64,
316}
317
318/// Proto representation of `DescribeWorkflowRequest`.
319#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
320pub struct ProtoDescribeWorkflowRequest {
321    /// Namespace that scopes the operation.
322    #[prost(string, tag = "1")]
323    pub namespace: String,
324    /// Target workflow identifier.
325    #[prost(message, optional, tag = "2")]
326    pub workflow_id: Option<ProtoWorkflowId>,
327    /// Target run identifier.
328    #[prost(message, optional, tag = "3")]
329    pub run_id: Option<ProtoRunId>,
330    /// Whether event history should be included in the response.
331    #[prost(bool, tag = "4")]
332    pub include_history: bool,
333}
334
335/// Proto representation of `DescribeWorkflowResponse`.
336#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
337pub struct ProtoDescribeWorkflowResponse {
338    /// Serde-encoded `aion_core::WorkflowSummary` envelope.
339    #[prost(message, optional, tag = "1")]
340    pub summary: Option<WireEnvelope>,
341    /// Optional serde-encoded `aion_core::Event` envelopes.
342    #[prost(message, repeated, tag = "2")]
343    pub history: Vec<WireEnvelope>,
344}
345
346#[cfg(test)]
347mod tests {
348    use std::collections::HashMap;
349
350    use aion_core::SearchAttributeValue;
351    use aion_store::visibility::{ListWorkflowsFilter, SearchAttributePredicate};
352    use chrono::{DateTime, Utc};
353    use prost::Message;
354    use serde::de::DeserializeOwned;
355    use serde_json::json;
356
357    use super::{
358        ProtoCountWorkflowsRequest, ProtoCountWorkflowsResponse, ProtoListWorkflowsRequest,
359        ProtoListWorkflowsResponse, ProtoQueryRequest, ProtoQueryResponse, ProtoReopenRequest,
360        ProtoReopenResponse, ProtoStartWorkflowRequest, ProtoStartWorkflowResponse,
361        proto_query_response,
362    };
363    use crate::convert::{
364        ProtoPayload, ProtoRunId, ProtoWorkflowId, decode_core_value, encode_core_value,
365    };
366    use crate::error::{ProtoWireError, WireError};
367
368    fn workflow_id() -> aion_core::WorkflowId {
369        aion_core::WorkflowId::new(uuid::Uuid::nil())
370    }
371
372    fn run_id() -> aion_core::RunId {
373        aion_core::RunId::new(uuid::Uuid::nil())
374    }
375
376    fn payload(label: &str) -> Result<ProtoPayload, aion_core::PayloadError> {
377        Ok(ProtoPayload::from(aion_core::Payload::from_json(
378            &json!({ "label": label }),
379        )?))
380    }
381
382    fn recorded_at() -> Result<DateTime<Utc>, chrono::ParseError> {
383        Ok(DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")?.with_timezone(&Utc))
384    }
385
386    fn assert_json_round_trip<T>(value: &T) -> Result<(), serde_json::Error>
387    where
388        T: Clone + PartialEq + serde::Serialize + DeserializeOwned,
389    {
390        let encoded = serde_json::to_string(value)?;
391        let decoded = serde_json::from_str::<T>(&encoded)?;
392        assert!(decoded == *value);
393        Ok(())
394    }
395
396    fn assert_proto_round_trip<T>(value: &T) -> Result<(), Box<dyn std::error::Error>>
397    where
398        T: Clone + PartialEq + Message + Default,
399    {
400        let mut bytes = Vec::new();
401        value.encode(&mut bytes)?;
402        let decoded = T::decode(bytes.as_slice())?;
403        assert!(decoded == *value);
404        Ok(())
405    }
406
407    #[test]
408    fn start_workflow_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
409        let request = ProtoStartWorkflowRequest {
410            namespace: String::from("tenant-a"),
411            workflow_type: String::from("checkout"),
412            input: Some(payload("input")?),
413            routing_key: Some(String::from("tenant-a/order-1")),
414            task_queue: Some(String::from("gpu")),
415            display_name: Some(String::from("Order 1 checkout")),
416        };
417        let response = ProtoStartWorkflowResponse {
418            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
419            run_id: Some(ProtoRunId::from(run_id())),
420        };
421
422        assert_json_round_trip(&request)?;
423        assert_proto_round_trip(&request)?;
424        assert_json_round_trip(&response)?;
425        assert_proto_round_trip(&response)?;
426        Ok(())
427    }
428
429    #[test]
430    fn list_workflows_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
431        let filter = ListWorkflowsFilter {
432            workflow_type: Some(String::from("checkout")),
433            status: Some(aion_core::WorkflowStatus::Running),
434            search_attributes: vec![SearchAttributePredicate::Equals {
435                name: String::from("customer_id"),
436                value: SearchAttributeValue::String(String::from("12345")),
437            }],
438            limit: Some(10),
439            offset: Some(5),
440            ..ListWorkflowsFilter::default()
441        };
442        let summary = aion_store::visibility::WorkflowSummary {
443            workflow_id: workflow_id(),
444            run_id: run_id(),
445            workflow_type: String::from("checkout"),
446            status: aion_core::WorkflowStatus::Running,
447            start_time: recorded_at()?,
448            close_time: None,
449            failed_step: None,
450            failure_reason: None,
451            search_attributes: HashMap::from([(
452                String::from("customer_id"),
453                SearchAttributeValue::String(String::from("12345")),
454            )]),
455        };
456        let filter_envelope = encode_core_value("tenant-a", Some(String::from("r1")), &filter)?;
457        let summary_envelope = encode_core_value("tenant-a", None, &summary)?;
458        let request = ProtoListWorkflowsRequest {
459            namespace: String::from("tenant-a"),
460            filter: Some(filter_envelope.clone()),
461        };
462        let response = ProtoListWorkflowsResponse {
463            summaries: vec![summary_envelope.clone()],
464        };
465        let count_request = ProtoCountWorkflowsRequest {
466            namespace: String::from("tenant-a"),
467            filter: Some(filter_envelope.clone()),
468        };
469        let count_response = ProtoCountWorkflowsResponse { count: 1 };
470
471        assert_json_round_trip(&request)?;
472        assert_proto_round_trip(&request)?;
473        assert_json_round_trip(&response)?;
474        assert_proto_round_trip(&response)?;
475        assert_json_round_trip(&count_request)?;
476        assert_proto_round_trip(&count_request)?;
477        assert_json_round_trip(&count_response)?;
478        assert_proto_round_trip(&count_response)?;
479        assert_eq!(
480            decode_core_value::<ListWorkflowsFilter>(&filter_envelope)?,
481            filter
482        );
483        assert_eq!(
484            decode_core_value::<aion_store::visibility::WorkflowSummary>(&summary_envelope)?,
485            summary
486        );
487        Ok(())
488    }
489
490    #[test]
491    fn query_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
492        let request = ProtoQueryRequest {
493            namespace: String::from("tenant-a"),
494            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
495            run_id: Some(ProtoRunId::from(run_id())),
496            query_name: String::from("state"),
497            arguments: Some(payload("arguments")?),
498        };
499        // A caller that supplies no arguments is a distinct encoded shape, not
500        // an error: the field is optional on the wire and the server supplies
501        // the canonical `null` document in its place.
502        let no_arguments_request = ProtoQueryRequest {
503            arguments: None,
504            ..request.clone()
505        };
506        let result_response = ProtoQueryResponse {
507            outcome: Some(proto_query_response::Outcome::Result(payload("result")?)),
508        };
509        let error_response = ProtoQueryResponse {
510            outcome: Some(proto_query_response::Outcome::Error(ProtoWireError::from(
511                WireError::unknown_query("state query is not registered"),
512            ))),
513        };
514
515        assert_json_round_trip(&request)?;
516        assert_proto_round_trip(&request)?;
517        assert_json_round_trip(&no_arguments_request)?;
518        assert_proto_round_trip(&no_arguments_request)?;
519        assert_json_round_trip(&result_response)?;
520        assert_proto_round_trip(&result_response)?;
521        assert_json_round_trip(&error_response)?;
522        assert_proto_round_trip(&error_response)?;
523        // The two shapes stay distinguishable across a proto round trip: a
524        // decoder can tell "no arguments supplied" from any supplied document.
525        assert_ne!(request, no_arguments_request);
526        Ok(())
527    }
528
529    /// #211: an unnamed start is a distinct wire shape from a named one, and
530    /// both round-trip across JSON and proto without conflating.
531    #[test]
532    fn start_workflow_display_name_absent_round_trips() -> Result<(), Box<dyn std::error::Error>> {
533        let named = ProtoStartWorkflowRequest {
534            namespace: String::from("tenant-a"),
535            workflow_type: String::from("checkout"),
536            input: Some(payload("input")?),
537            routing_key: None,
538            task_queue: None,
539            display_name: Some(String::from("Order 1 checkout")),
540        };
541        let unnamed = ProtoStartWorkflowRequest {
542            display_name: None,
543            ..named.clone()
544        };
545
546        assert_json_round_trip(&named)?;
547        assert_proto_round_trip(&named)?;
548        assert_json_round_trip(&unnamed)?;
549        assert_proto_round_trip(&unnamed)?;
550        // The two shapes stay distinguishable across the wire: a decoder can
551        // tell "no name supplied" from any supplied name.
552        assert_ne!(named, unnamed);
553        Ok(())
554    }
555
556    /// #211: the HAND-WRITTEN `display_name` and the GENERATED stub's
557    /// `display_name` are the same wire field.
558    ///
559    /// Both sides declare tag 6, but "both say 6" is two claims, not agreement.
560    /// This encodes with one and decodes with the other, in both directions, so
561    /// a tag or wire-type drift between the `.proto` and the hand-written prost
562    /// derive is a failure here rather than a field that silently vanishes at a
563    /// real transport boundary. Compiled only under `generated`, which is the
564    /// only posture in which the stubs exist at all.
565    #[cfg(feature = "generated")]
566    #[test]
567    fn start_workflow_display_name_is_the_same_wire_field_as_the_generated_stub()
568    -> Result<(), Box<dyn std::error::Error>> {
569        const NAME: &str = "Nightly settlement";
570        let hand_written = ProtoStartWorkflowRequest {
571            namespace: String::from("tenant-a"),
572            workflow_type: String::from("checkout"),
573            input: None,
574            routing_key: None,
575            task_queue: None,
576            display_name: Some(String::from(NAME)),
577        };
578
579        // Hand-written -> generated.
580        let mut bytes = Vec::new();
581        hand_written.encode(&mut bytes)?;
582        let decoded = crate::generated::StartWorkflowRequest::decode(bytes.as_slice())?;
583        assert_eq!(
584            decoded.display_name.as_deref(),
585            Some(NAME),
586            "the generated stub must read the hand-written display_name"
587        );
588
589        // Generated -> hand-written.
590        let mut bytes = Vec::new();
591        decoded.encode(&mut bytes)?;
592        let round_tripped = ProtoStartWorkflowRequest::decode(bytes.as_slice())?;
593        assert_eq!(round_tripped, hand_written);
594
595        // And the field key itself is pinned: tag 6, length-delimited, is
596        // (6 << 3) | 2 = 0x32. An unnamed start emits it nowhere.
597        let mut bytes = Vec::new();
598        ProtoStartWorkflowRequest {
599            namespace: String::new(),
600            workflow_type: String::new(),
601            input: None,
602            routing_key: None,
603            task_queue: None,
604            display_name: Some(String::from("x")),
605        }
606        .encode(&mut bytes)?;
607        assert_eq!(bytes, vec![0x32, 0x01, b'x']);
608
609        let mut bytes = Vec::new();
610        ProtoStartWorkflowRequest {
611            namespace: String::new(),
612            workflow_type: String::new(),
613            input: None,
614            routing_key: None,
615            task_queue: None,
616            display_name: None,
617        }
618        .encode(&mut bytes)?;
619        assert!(
620            bytes.is_empty(),
621            "an unnamed start must put nothing on the wire, got {bytes:?}"
622        );
623        Ok(())
624    }
625
626    /// #211: the hand-written Rename messages agree with the generated stubs,
627    /// in both directions.
628    #[cfg(feature = "generated")]
629    #[test]
630    fn rename_messages_agree_with_the_generated_stubs() -> Result<(), Box<dyn std::error::Error>> {
631        let request = super::ProtoRenameRequest {
632            namespace: String::from("tenant-a"),
633            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
634            run_id: Some(ProtoRunId::from(run_id())),
635            display_name: String::from("Nightly settlement"),
636        };
637        let mut bytes = Vec::new();
638        request.encode(&mut bytes)?;
639        let decoded = crate::generated::RenameRequest::decode(bytes.as_slice())?;
640        assert_eq!(decoded.display_name, "Nightly settlement");
641        assert_eq!(decoded.namespace, "tenant-a");
642        let mut bytes = Vec::new();
643        decoded.encode(&mut bytes)?;
644        assert_eq!(
645            super::ProtoRenameRequest::decode(bytes.as_slice())?,
646            request
647        );
648
649        let response = super::ProtoRenameResponse {
650            run_id: Some(ProtoRunId::from(run_id())),
651            display_name: String::from("Nightly settlement"),
652        };
653        let mut bytes = Vec::new();
654        response.encode(&mut bytes)?;
655        let decoded = crate::generated::RenameResponse::decode(bytes.as_slice())?;
656        assert_eq!(decoded.display_name, "Nightly settlement");
657        let mut bytes = Vec::new();
658        decoded.encode(&mut bytes)?;
659        assert_eq!(
660            super::ProtoRenameResponse::decode(bytes.as_slice())?,
661            response
662        );
663        Ok(())
664    }
665
666    #[test]
667    fn rename_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
668        let request = super::ProtoRenameRequest {
669            namespace: String::from("tenant-a"),
670            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
671            run_id: Some(ProtoRunId::from(run_id())),
672            display_name: String::from("Nightly settlement"),
673        };
674        let response = super::ProtoRenameResponse {
675            run_id: Some(ProtoRunId::from(run_id())),
676            display_name: String::from("Nightly settlement"),
677        };
678
679        assert_json_round_trip(&request)?;
680        assert_proto_round_trip(&request)?;
681        assert_json_round_trip(&response)?;
682        assert_proto_round_trip(&response)?;
683        Ok(())
684    }
685
686    #[test]
687    fn reopen_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
688        let request = ProtoReopenRequest {
689            namespace: String::from("tenant-a"),
690            workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
691            run_id: Some(ProtoRunId::from(run_id())),
692        };
693        let response = ProtoReopenResponse {
694            run_id: Some(ProtoRunId::from(run_id())),
695            status: crate::convert::ProtoWorkflowStatus::Running as i32,
696        };
697
698        assert_json_round_trip(&request)?;
699        assert_proto_round_trip(&request)?;
700        assert_json_round_trip(&response)?;
701        assert_proto_round_trip(&response)?;
702        Ok(())
703    }
704}
aion_proto/workflow.rs

aion_proto/
workflow.rs
