tmprl_client/ops/
history.rs

1//! Reading a workflow history, and flattening it into normalised events.
2//!
3//! The match over the attributes `oneof` is **exhaustive on purpose**, per design rule 4.
4//! Temporal adds event types regularly: Nexus, worker versioning and workflow pausing are
5//! all recent. With a `_ => {}` arm a new event type renders as a blank row and nobody
6//! notices for a release or two; exhaustive, it is a compile error the moment the protos
7//! are bumped, which is exactly when we want to hear about it.
8//!
9//! Grouping keys come straight from the protocol's back-references. Which field to follow
10//! is not uniform, so it is spelled out per arm rather than guessed:
11//!
12//! * activities, workflow tasks and Nexus operations → `scheduled_event_id`
13//! * child and external workflows → `initiated_event_id`
14//! * timers → `started_event_id` (the id of the `TimerStarted` event)
15//! * updates → `accepted_event_id`
16//!
17//! `workflow_task_completed_event_id` appears on many of these too, but it points at the
18//! workflow task that *caused* the command, not at the thing's own group, following it
19//! would file every activity under the task that scheduled it.
20
21use std::collections::HashMap;
22
23use temporalio_client::tonic::Request;
24use temporalio_common::protos::temporal::api::{
25    common::v1::{Payload as ProtoPayload, Payloads, WorkflowExecution},
26    enums::v1::HistoryEventFilterType,
27    failure::v1::Failure,
28    history::v1::{HistoryEvent, history_event::Attributes},
29    update::v1::outcome::Value as OutcomeValue,
30    workflowservice::v1::GetWorkflowExecutionHistoryRequest,
31};
32use tmprl_core::history::{Category, GroupRef, NormalizedEvent, Outcome, Role};
33use tmprl_core::payload::Payload;
34
35use super::OpError;
36use crate::Conn;
37
38/// One page of a workflow's history.
39#[derive(Debug, Clone, Default)]
40pub struct HistoryPage {
41    pub events: Vec<NormalizedEvent>,
42    /// Empty on the last page.
43    pub next_page_token: Vec<u8>,
44}
45
46impl HistoryPage {
47    pub fn has_more(&self) -> bool {
48        !self.next_page_token.is_empty()
49    }
50}
51
52impl Conn {
53    /// One long-poll step of follow mode.
54    ///
55    /// This is [`Conn::get_history`] with `wait_new_event: true`, which makes the call
56    /// **block until the workflow does something**, up to about a minute, then it returns
57    /// empty-handed and you call again. Never reach for this outside a task dedicated to
58    /// following; anything else it is on will simply stop.
59    ///
60    /// The behaviour of the continuation token differs from paging, and follow mode is built
61    /// on the difference. Measured against a dev server:
62    ///
63    /// | | `wait_new_event: false` | `wait_new_event: true` |
64    /// |---|---|---|
65    /// | running workflow, caught up | returns 0 events, **empty** token | blocks, then returns new events, token stays non-empty |
66    /// | closed workflow | empty token | empty token, terminal event last |
67    ///
68    /// So an **empty token here means the workflow has closed** and there is nothing further
69    /// to follow: the loop's termination condition, and it is authoritative in a way
70    /// that inspecting the last event's type is not.
71    ///
72    /// Passing an empty token restarts from event 1, so a caller resuming a follow should
73    /// hand back the last non-empty token it saw. The page that token sits in is replayed,
74    /// which is why events are merged rather than appended, see
75    /// `tmprl_core::history::merge_events`.
76    pub async fn follow_history(
77        &self,
78        namespace: &str,
79        workflow_id: &str,
80        run_id: &str,
81        next_page_token: Vec<u8>,
82    ) -> Result<HistoryPage, OpError> {
83        self.history_request(namespace, workflow_id, run_id, 100, next_page_token, true)
84            .await
85    }
86
87    /// One page of history, normalised.
88    ///
89    /// `wait_new_event` is false here. Setting it true turns this into a long poll that does
90    /// not return until something happens, which is correct for follow mode and a hang
91    /// everywhere else, so follow mode gets [`Conn::follow_history`] rather than a flag on
92    /// this one that is easy to pass by accident.
93    pub async fn get_history(
94        &self,
95        namespace: &str,
96        workflow_id: &str,
97        run_id: &str,
98        page_size: i32,
99        next_page_token: Vec<u8>,
100    ) -> Result<HistoryPage, OpError> {
101        self.history_request(
102            namespace,
103            workflow_id,
104            run_id,
105            page_size,
106            next_page_token,
107            false,
108        )
109        .await
110    }
111
112    async fn history_request(
113        &self,
114        namespace: &str,
115        workflow_id: &str,
116        run_id: &str,
117        page_size: i32,
118        next_page_token: Vec<u8>,
119        wait_new_event: bool,
120    ) -> Result<HistoryPage, OpError> {
121        let resp = self
122            .wf()
123            .get_workflow_execution_history(Request::new(GetWorkflowExecutionHistoryRequest {
124                namespace: namespace.to_string(),
125                execution: Some(WorkflowExecution {
126                    workflow_id: workflow_id.to_string(),
127                    run_id: run_id.to_string(),
128                }),
129                maximum_page_size: page_size,
130                next_page_token,
131                wait_new_event,
132                history_event_filter_type: HistoryEventFilterType::AllEvent as i32,
133                skip_archival: false,
134            }))
135            .await
136            .map_err(|s| OpError::rpc("GetWorkflowExecutionHistory", s))?
137            .into_inner();
138
139        Ok(HistoryPage {
140            events: resp
141                .history
142                .map(|h| h.events)
143                .unwrap_or_default()
144                .into_iter()
145                .map(normalize)
146                .collect(),
147            next_page_token: resp.next_page_token,
148        })
149    }
150}
151
152/// What the match over the attributes produces. Assembled into a [`NormalizedEvent`] with
153/// the id and timestamp, which every event has regardless of its type.
154struct Mapped {
155    category: Category,
156    group: GroupRef,
157    role: Role,
158    outcome: Outcome,
159    subject: String,
160    attempt: Option<i32>,
161    failure: Option<String>,
162    fields: Vec<(&'static str, String)>,
163    payloads: Vec<(String, Payload)>,
164}
165
166/// Start an arm. `group` is the group this event joins; `role` is what it does to it.
167fn at(category: Category, group: GroupRef, role: Role) -> Mapped {
168    Mapped {
169        category,
170        group,
171        role,
172        outcome: Outcome::Pending,
173        subject: String::new(),
174        attempt: None,
175        failure: None,
176        fields: Vec::new(),
177        payloads: Vec::new(),
178    }
179}
180
181impl Mapped {
182    fn subject(mut self, s: impl Into<String>) -> Self {
183        self.subject = s.into();
184        self
185    }
186    fn ends(mut self, outcome: Outcome) -> Self {
187        self.outcome = outcome;
188        self
189    }
190    fn failed(mut self, f: Option<Failure>) -> Self {
191        self.failure = f.map(|f| f.message);
192        self
193    }
194    fn attempt(mut self, n: i32) -> Self {
195        self.attempt = Some(n);
196        self
197    }
198    fn field(mut self, k: &'static str, v: impl Into<String>) -> Self {
199        let v = v.into();
200        if !v.is_empty() {
201            self.fields.push((k, v));
202        }
203        self
204    }
205
206    /// Attach an argument list. A single value is labelled plainly; several are indexed,
207    /// because an activity's third argument is not interchangeable with its first.
208    fn args(mut self, label: &str, p: Option<Payloads>) -> Self {
209        let Some(list) = p else { return self };
210        let n = list.payloads.len();
211        for (i, raw) in list.payloads.into_iter().enumerate() {
212            let name = if n == 1 {
213                label.to_string()
214            } else {
215                format!("{label}[{i}]")
216            };
217            self.payloads.push((name, convert(raw)));
218        }
219        self
220    }
221
222    /// Attach a keyed set of argument lists, as a marker records.
223    ///
224    /// The map has no order on the wire, so the keys are sorted: a detail pane that reshuffles
225    /// between refreshes cannot be read.
226    fn keyed_args(mut self, map: HashMap<String, Payloads>) -> Self {
227        let mut keys: Vec<String> = map.keys().cloned().collect();
228        keys.sort();
229        for k in keys {
230            let list = map[&k].clone();
231            self = self.args(&k, Some(list));
232        }
233        self
234    }
235
236    /// Attach a single payload.
237    fn arg(mut self, label: &str, p: Option<ProtoPayload>) -> Self {
238        if let Some(raw) = p {
239            self.payloads.push((label.to_string(), convert(raw)));
240        }
241        self
242    }
243}
244
245/// Protobuf payload to domain payload. The metadata values are bytes on the wire; the two
246/// keys tmprl reads are ASCII.
247fn convert(p: ProtoPayload) -> Payload {
248    let meta = |k: &str| {
249        p.metadata
250            .get(k)
251            .and_then(|v| std::str::from_utf8(v).ok())
252            .map(str::to_string)
253    };
254    Payload {
255        encoding: meta("encoding").unwrap_or_default(),
256        type_hint: meta("type"),
257        data: p.data,
258    }
259}
260
261/// Flatten one protobuf event.
262pub fn normalize(e: HistoryEvent) -> NormalizedEvent {
263    let id = e.event_id;
264    let time = e
265        .event_time
266        .map(|t| t.seconds * 1000 + i64::from(t.nanos) / 1_000_000);
267    // `event_type()` borrows, so read the name before the attributes are moved out.
268    let name = event_name(e.event_type().as_str_name());
269
270    let m = match e.attributes {
271        // An event with no attributes is representable on the wire but meaningless. It is
272        // rendered as itself rather than dropped, so a history never silently loses a row.
273        None => at(Category::Workflow, GroupRef::Workflow, Role::Continues),
274
275        Some(Attributes::WorkflowExecutionStartedEventAttributes(a)) => {
276            at(Category::Workflow, GroupRef::Workflow, Role::Opens)
277                .subject(a.workflow_type.map(|t| t.name).unwrap_or_default())
278                .field(
279                    "taskQueue",
280                    a.task_queue.map(|q| q.name).unwrap_or_default(),
281                )
282                .field("attempt", a.attempt.to_string())
283                .field("firstRunId", a.first_execution_run_id)
284                .args("input", a.input)
285                .args("lastCompletionResult", a.last_completion_result)
286        }
287        Some(Attributes::WorkflowExecutionCompletedEventAttributes(a)) => {
288            at(Category::Workflow, GroupRef::Workflow, Role::Closes)
289                .ends(Outcome::Completed)
290                .args("result", a.result)
291        }
292        Some(Attributes::WorkflowExecutionFailedEventAttributes(a)) => {
293            at(Category::Workflow, GroupRef::Workflow, Role::Closes)
294                .ends(Outcome::Failed)
295                .failed(a.failure)
296        }
297        Some(Attributes::WorkflowExecutionTimedOutEventAttributes(_)) => {
298            at(Category::Workflow, GroupRef::Workflow, Role::Closes).ends(Outcome::TimedOut)
299        }
300        Some(Attributes::WorkflowExecutionCanceledEventAttributes(a)) => {
301            at(Category::Workflow, GroupRef::Workflow, Role::Closes)
302                .ends(Outcome::Canceled)
303                .args("details", a.details)
304        }
305        Some(Attributes::WorkflowExecutionTerminatedEventAttributes(a)) => {
306            at(Category::Workflow, GroupRef::Workflow, Role::Closes)
307                .ends(Outcome::Terminated)
308                .field("reason", a.reason)
309                .field("identity", a.identity)
310                .args("details", a.details)
311        }
312        Some(Attributes::WorkflowExecutionContinuedAsNewEventAttributes(a)) => {
313            at(Category::Workflow, GroupRef::Workflow, Role::Closes)
314                .ends(Outcome::ContinuedAsNew)
315                .field("newRunId", a.new_execution_run_id)
316                .args("input", a.input)
317                .args("lastCompletionResult", a.last_completion_result)
318        }
319        Some(Attributes::WorkflowExecutionCancelRequestedEventAttributes(a)) => {
320            at(Category::Workflow, GroupRef::Workflow, Role::Continues)
321                .field("cause", a.cause)
322                .field("identity", a.identity)
323        }
324        Some(Attributes::WorkflowExecutionSignaledEventAttributes(a)) => {
325            at(Category::Workflow, GroupRef::Workflow, Role::Continues)
326                .subject(a.signal_name)
327                .field("identity", a.identity)
328                .args("input", a.input)
329        }
330        Some(Attributes::WorkflowExecutionPausedEventAttributes(_)) => {
331            at(Category::Workflow, GroupRef::Workflow, Role::Continues)
332        }
333        Some(Attributes::WorkflowExecutionUnpausedEventAttributes(_)) => {
334            at(Category::Workflow, GroupRef::Workflow, Role::Continues)
335        }
336        Some(Attributes::WorkflowExecutionOptionsUpdatedEventAttributes(_)) => {
337            at(Category::Workflow, GroupRef::Workflow, Role::Continues)
338        }
339        Some(Attributes::WorkflowPropertiesModifiedEventAttributes(_)) => {
340            at(Category::Workflow, GroupRef::Workflow, Role::Continues)
341        }
342        Some(Attributes::WorkflowPropertiesModifiedExternallyEventAttributes(_)) => {
343            at(Category::Workflow, GroupRef::Workflow, Role::Continues)
344        }
345        Some(Attributes::WorkflowExecutionTimeSkippingTransitionedEventAttributes(_)) => {
346            at(Category::Workflow, GroupRef::Workflow, Role::Continues)
347        }
348
349        Some(Attributes::WorkflowTaskScheduledEventAttributes(a)) => {
350            at(Category::WorkflowTask, GroupRef::Opened(id), Role::Opens)
351                .attempt(a.attempt)
352                .field(
353                    "taskQueue",
354                    a.task_queue.map(|q| q.name).unwrap_or_default(),
355                )
356        }
357        Some(Attributes::WorkflowTaskStartedEventAttributes(a)) => at(
358            Category::WorkflowTask,
359            GroupRef::Opened(a.scheduled_event_id),
360            Role::Continues,
361        )
362        .field("identity", a.identity),
363        Some(Attributes::WorkflowTaskCompletedEventAttributes(a)) => at(
364            Category::WorkflowTask,
365            GroupRef::Opened(a.scheduled_event_id),
366            Role::Closes,
367        )
368        .ends(Outcome::Completed),
369        Some(Attributes::WorkflowTaskTimedOutEventAttributes(a)) => at(
370            Category::WorkflowTask,
371            GroupRef::Opened(a.scheduled_event_id),
372            Role::Closes,
373        )
374        .ends(Outcome::TimedOut),
375        Some(Attributes::WorkflowTaskFailedEventAttributes(a)) => at(
376            Category::WorkflowTask,
377            GroupRef::Opened(a.scheduled_event_id),
378            Role::Closes,
379        )
380        .ends(Outcome::Failed)
381        .failed(a.failure),
382
383        Some(Attributes::ActivityTaskScheduledEventAttributes(a)) => {
384            at(Category::Activity, GroupRef::Opened(id), Role::Opens)
385                .subject(a.activity_type.map(|t| t.name).unwrap_or_default())
386                .field("activityId", a.activity_id)
387                .field(
388                    "taskQueue",
389                    a.task_queue.map(|q| q.name).unwrap_or_default(),
390                )
391                .args("input", a.input)
392        }
393        Some(Attributes::ActivityTaskStartedEventAttributes(a)) => at(
394            Category::Activity,
395            GroupRef::Opened(a.scheduled_event_id),
396            Role::Continues,
397        )
398        // A retry does not schedule again; this is where the attempt count lives, and
399        // `last_failure` is why the previous attempt did not stick.
400        .attempt(a.attempt)
401        .failed(a.last_failure)
402        .field("identity", a.identity),
403        Some(Attributes::ActivityTaskCompletedEventAttributes(a)) => at(
404            Category::Activity,
405            GroupRef::Opened(a.scheduled_event_id),
406            Role::Closes,
407        )
408        .ends(Outcome::Completed)
409        .args("result", a.result),
410        Some(Attributes::ActivityTaskFailedEventAttributes(a)) => at(
411            Category::Activity,
412            GroupRef::Opened(a.scheduled_event_id),
413            Role::Closes,
414        )
415        .ends(Outcome::Failed)
416        .failed(a.failure),
417        Some(Attributes::ActivityTaskTimedOutEventAttributes(a)) => at(
418            Category::Activity,
419            GroupRef::Opened(a.scheduled_event_id),
420            Role::Closes,
421        )
422        .ends(Outcome::TimedOut)
423        .failed(a.failure),
424        Some(Attributes::ActivityTaskCanceledEventAttributes(a)) => at(
425            Category::Activity,
426            GroupRef::Opened(a.scheduled_event_id),
427            Role::Closes,
428        )
429        .ends(Outcome::Canceled)
430        .args("details", a.details),
431        Some(Attributes::ActivityTaskCancelRequestedEventAttributes(a)) => at(
432            Category::Activity,
433            GroupRef::Opened(a.scheduled_event_id),
434            Role::Continues,
435        ),
436        Some(Attributes::ActivityPropertiesModifiedExternallyEventAttributes(a)) => at(
437            Category::Activity,
438            GroupRef::Opened(a.scheduled_event_id),
439            Role::Continues,
440        ),
441
442        Some(Attributes::TimerStartedEventAttributes(a)) => {
443            at(Category::Timer, GroupRef::Opened(id), Role::Opens)
444                .subject(a.timer_id)
445                .field(
446                    "startToFireTimeout",
447                    a.start_to_fire_timeout
448                        .map(|d| format!("{}s", d.seconds))
449                        .unwrap_or_default(),
450                )
451        }
452        Some(Attributes::TimerFiredEventAttributes(a)) => at(
453            Category::Timer,
454            GroupRef::Opened(a.started_event_id),
455            Role::Closes,
456        )
457        .ends(Outcome::Completed),
458        Some(Attributes::TimerCanceledEventAttributes(a)) => at(
459            Category::Timer,
460            GroupRef::Opened(a.started_event_id),
461            Role::Closes,
462        )
463        .ends(Outcome::Canceled),
464
465        Some(Attributes::StartChildWorkflowExecutionInitiatedEventAttributes(a)) => {
466            at(Category::ChildWorkflow, GroupRef::Opened(id), Role::Opens)
467                .subject(a.workflow_type.map(|t| t.name).unwrap_or_default())
468                .field("workflowId", a.workflow_id)
469                .field("namespace", a.namespace)
470                .args("input", a.input)
471        }
472        Some(Attributes::StartChildWorkflowExecutionFailedEventAttributes(a)) => at(
473            Category::ChildWorkflow,
474            GroupRef::Opened(a.initiated_event_id),
475            Role::Closes,
476        )
477        .ends(Outcome::Failed)
478        .field("workflowId", a.workflow_id),
479        Some(Attributes::ChildWorkflowExecutionStartedEventAttributes(a)) => at(
480            Category::ChildWorkflow,
481            GroupRef::Opened(a.initiated_event_id),
482            Role::Continues,
483        )
484        .field(
485            "runId",
486            a.workflow_execution.map(|w| w.run_id).unwrap_or_default(),
487        ),
488        Some(Attributes::ChildWorkflowExecutionCompletedEventAttributes(a)) => at(
489            Category::ChildWorkflow,
490            GroupRef::Opened(a.initiated_event_id),
491            Role::Closes,
492        )
493        .ends(Outcome::Completed)
494        .args("result", a.result),
495        Some(Attributes::ChildWorkflowExecutionFailedEventAttributes(a)) => at(
496            Category::ChildWorkflow,
497            GroupRef::Opened(a.initiated_event_id),
498            Role::Closes,
499        )
500        .ends(Outcome::Failed)
501        .failed(a.failure),
502        Some(Attributes::ChildWorkflowExecutionCanceledEventAttributes(a)) => at(
503            Category::ChildWorkflow,
504            GroupRef::Opened(a.initiated_event_id),
505            Role::Closes,
506        )
507        .ends(Outcome::Canceled)
508        .args("details", a.details),
509        Some(Attributes::ChildWorkflowExecutionTimedOutEventAttributes(a)) => at(
510            Category::ChildWorkflow,
511            GroupRef::Opened(a.initiated_event_id),
512            Role::Closes,
513        )
514        .ends(Outcome::TimedOut),
515        Some(Attributes::ChildWorkflowExecutionTerminatedEventAttributes(a)) => at(
516            Category::ChildWorkflow,
517            GroupRef::Opened(a.initiated_event_id),
518            Role::Closes,
519        )
520        .ends(Outcome::Terminated),
521
522        Some(Attributes::SignalExternalWorkflowExecutionInitiatedEventAttributes(a)) => at(
523            Category::ExternalWorkflow,
524            GroupRef::Opened(id),
525            Role::Opens,
526        )
527        .subject(a.signal_name)
528        .field(
529            "workflowId",
530            a.workflow_execution
531                .map(|w| w.workflow_id)
532                .unwrap_or_default(),
533        )
534        .field("namespace", a.namespace)
535        .args("input", a.input),
536        Some(Attributes::SignalExternalWorkflowExecutionFailedEventAttributes(a)) => at(
537            Category::ExternalWorkflow,
538            GroupRef::Opened(a.initiated_event_id),
539            Role::Closes,
540        )
541        .ends(Outcome::Failed)
542        .field("cause", a.cause.to_string()),
543        Some(Attributes::ExternalWorkflowExecutionSignaledEventAttributes(a)) => at(
544            Category::ExternalWorkflow,
545            GroupRef::Opened(a.initiated_event_id),
546            Role::Closes,
547        )
548        .ends(Outcome::Completed),
549        Some(Attributes::RequestCancelExternalWorkflowExecutionInitiatedEventAttributes(a)) => at(
550            Category::ExternalWorkflow,
551            GroupRef::Opened(id),
552            Role::Opens,
553        )
554        .subject("cancel")
555        .field(
556            "workflowId",
557            a.workflow_execution
558                .map(|w| w.workflow_id)
559                .unwrap_or_default(),
560        ),
561        Some(Attributes::RequestCancelExternalWorkflowExecutionFailedEventAttributes(a)) => at(
562            Category::ExternalWorkflow,
563            GroupRef::Opened(a.initiated_event_id),
564            Role::Closes,
565        )
566        .ends(Outcome::Failed)
567        .field("cause", a.cause.to_string()),
568        Some(Attributes::ExternalWorkflowExecutionCancelRequestedEventAttributes(a)) => at(
569            Category::ExternalWorkflow,
570            GroupRef::Opened(a.initiated_event_id),
571            Role::Closes,
572        )
573        .ends(Outcome::Completed),
574
575        Some(Attributes::WorkflowExecutionUpdateAdmittedEventAttributes(_)) => {
576            at(Category::Update, GroupRef::Opened(id), Role::Opens)
577        }
578        Some(Attributes::WorkflowExecutionUpdateAcceptedEventAttributes(a)) => {
579            // The arguments live two levels down, on the request the acceptance echoes back.
580            let input = a
581                .accepted_request
582                .and_then(|r| r.input)
583                .and_then(|i| i.args);
584            at(Category::Update, GroupRef::Opened(id), Role::Opens)
585                .subject(a.protocol_instance_id)
586                .args("input", input)
587        }
588        Some(Attributes::WorkflowExecutionUpdateCompletedEventAttributes(a)) => {
589            // An update can be accepted and then fail, and the outcome is where that shows.
590            // Reading only the event type would report every such update as completed.
591            let m = at(
592                Category::Update,
593                GroupRef::Opened(a.accepted_event_id),
594                Role::Closes,
595            );
596            match a.outcome.and_then(|o| o.value) {
597                Some(OutcomeValue::Success(p)) => {
598                    m.ends(Outcome::Completed).args("result", Some(p))
599                }
600                Some(OutcomeValue::Failure(f)) => m.ends(Outcome::Failed).failed(Some(f)),
601                None => m.ends(Outcome::Completed),
602            }
603        }
604        Some(Attributes::WorkflowExecutionUpdateRejectedEventAttributes(a)) => {
605            // Rejected before acceptance, so there is no accepted event to hang it on.
606            at(Category::Update, GroupRef::Opened(id), Role::Opens)
607                .subject(a.protocol_instance_id)
608                .ends(Outcome::Rejected)
609                .failed(a.failure)
610        }
611
612        Some(Attributes::NexusOperationScheduledEventAttributes(a)) => {
613            at(Category::Nexus, GroupRef::Opened(id), Role::Opens)
614                .subject(format!("{}/{}", a.service, a.operation))
615                .field("endpoint", a.endpoint)
616                .arg("input", a.input)
617        }
618        Some(Attributes::NexusOperationStartedEventAttributes(a)) => at(
619            Category::Nexus,
620            GroupRef::Opened(a.scheduled_event_id),
621            Role::Continues,
622        ),
623        Some(Attributes::NexusOperationCompletedEventAttributes(a)) => at(
624            Category::Nexus,
625            GroupRef::Opened(a.scheduled_event_id),
626            Role::Closes,
627        )
628        .ends(Outcome::Completed)
629        .arg("result", a.result),
630        Some(Attributes::NexusOperationFailedEventAttributes(a)) => at(
631            Category::Nexus,
632            GroupRef::Opened(a.scheduled_event_id),
633            Role::Closes,
634        )
635        .ends(Outcome::Failed)
636        .failed(a.failure),
637        Some(Attributes::NexusOperationCanceledEventAttributes(a)) => at(
638            Category::Nexus,
639            GroupRef::Opened(a.scheduled_event_id),
640            Role::Closes,
641        )
642        .ends(Outcome::Canceled),
643        Some(Attributes::NexusOperationTimedOutEventAttributes(a)) => at(
644            Category::Nexus,
645            GroupRef::Opened(a.scheduled_event_id),
646            Role::Closes,
647        )
648        .ends(Outcome::TimedOut),
649        Some(Attributes::NexusOperationCancelRequestedEventAttributes(a)) => at(
650            Category::Nexus,
651            GroupRef::Opened(a.scheduled_event_id),
652            Role::Continues,
653        ),
654        Some(Attributes::NexusOperationCancelRequestCompletedEventAttributes(a)) => at(
655            Category::Nexus,
656            GroupRef::Opened(a.scheduled_event_id),
657            Role::Continues,
658        ),
659        Some(Attributes::NexusOperationCancelRequestFailedEventAttributes(a)) => at(
660            Category::Nexus,
661            GroupRef::Opened(a.scheduled_event_id),
662            Role::Continues,
663        )
664        .failed(a.failure),
665
666        Some(Attributes::MarkerRecordedEventAttributes(a)) => {
667            at(Category::Marker, GroupRef::Opened(id), Role::Opens)
668                .subject(a.marker_name)
669                .failed(a.failure)
670                .keyed_args(a.details)
671        }
672        Some(Attributes::UpsertWorkflowSearchAttributesEventAttributes(_)) => at(
673            Category::SearchAttributes,
674            GroupRef::Opened(id),
675            Role::Opens,
676        ),
677    };
678
679    NormalizedEvent {
680        id,
681        time,
682        name,
683        category: m.category,
684        group: m.group,
685        role: m.role,
686        outcome: m.outcome,
687        subject: m.subject,
688        attempt: m.attempt,
689        failure: m.failure,
690        fields: m.fields,
691        payloads: m.payloads,
692    }
693}
694
695/// `EVENT_TYPE_ACTIVITY_TASK_SCHEDULED` → `ActivityTaskScheduled`.
696///
697/// Returns a `&'static str` by matching the protobuf's own `as_str_name()` output, which is
698/// itself `&'static`. Building the name at runtime would mean leaking or allocating for
699/// every event in a history that can run to millions.
700fn event_name(proto: &'static str) -> &'static str {
701    // Trim the prefix and rebuild the CamelCase name is not possible without allocating, so
702    // the raw protobuf name is kept when it is not one we recognise. Unknown names only
703    // appear for event types added after this was written, which the exhaustive match above
704    // will have already flagged at compile time.
705    proto.strip_prefix("EVENT_TYPE_").unwrap_or(proto)
706}
707
708#[cfg(test)]
709mod tests {
710    use super::*;
711    use temporalio_common::protos::temporal::api::{
712        common::v1::ActivityType,
713        enums::v1::EventType,
714        history::v1::{
715            ActivityTaskFailedEventAttributes, ActivityTaskScheduledEventAttributes,
716            ActivityTaskStartedEventAttributes, MarkerRecordedEventAttributes,
717            TimerFiredEventAttributes, TimerStartedEventAttributes,
718            WorkflowExecutionCompletedEventAttributes, WorkflowExecutionSignaledEventAttributes,
719            WorkflowExecutionStartedEventAttributes,
720            WorkflowExecutionUpdateCompletedEventAttributes,
721        },
722        update::v1::Outcome as UpdateOutcome,
723    };
724
725    fn event(id: i64, ty: EventType, attrs: Attributes) -> HistoryEvent {
726        HistoryEvent {
727            event_id: id,
728            event_time: Some(prost_wkt_types::Timestamp {
729                seconds: id,
730                nanos: 0,
731            }),
732            event_type: ty as i32,
733            attributes: Some(attrs),
734            ..Default::default()
735        }
736    }
737
738    #[test]
739    fn an_activity_scheduled_event_opens_a_group_at_its_own_id() {
740        let n = normalize(event(
741            5,
742            EventType::ActivityTaskScheduled,
743            Attributes::ActivityTaskScheduledEventAttributes(
744                ActivityTaskScheduledEventAttributes {
745                    activity_id: "charge".into(),
746                    activity_type: Some(ActivityType {
747                        name: "ChargeCard".into(),
748                    }),
749                    // Points at the workflow task that scheduled this. Following it would
750                    // file the activity under that task instead of giving it its own group.
751                    workflow_task_completed_event_id: 4,
752                    ..Default::default()
753                },
754            ),
755        ));
756
757        assert_eq!(n.id, 5);
758        assert_eq!(n.group, GroupRef::Opened(5), "not Opened(4)");
759        assert_eq!(n.role, Role::Opens);
760        assert_eq!(n.category, Category::Activity);
761        assert_eq!(n.subject, "ChargeCard");
762        assert_eq!(n.name, "ACTIVITY_TASK_SCHEDULED");
763        assert_eq!(n.time, Some(5_000));
764        assert!(
765            n.fields
766                .iter()
767                .any(|(k, v)| *k == "activityId" && v == "charge")
768        );
769    }
770
771    #[test]
772    fn an_activity_started_event_joins_its_scheduled_group_and_carries_the_attempt() {
773        let n = normalize(event(
774            6,
775            EventType::ActivityTaskStarted,
776            Attributes::ActivityTaskStartedEventAttributes(ActivityTaskStartedEventAttributes {
777                scheduled_event_id: 5,
778                attempt: 3,
779                last_failure: Some(Failure {
780                    message: "card declined".into(),
781                    ..Default::default()
782                }),
783                ..Default::default()
784            }),
785        ));
786
787        assert_eq!(n.group, GroupRef::Opened(5));
788        assert_eq!(n.role, Role::Continues);
789        assert_eq!(
790            n.attempt,
791            Some(3),
792            "retries do not re-schedule; the count is here"
793        );
794        assert_eq!(n.failure.as_deref(), Some("card declined"));
795    }
796
797    #[test]
798    fn a_failed_activity_closes_its_group_with_the_failure_message() {
799        let n = normalize(event(
800            7,
801            EventType::ActivityTaskFailed,
802            Attributes::ActivityTaskFailedEventAttributes(ActivityTaskFailedEventAttributes {
803                scheduled_event_id: 5,
804                started_event_id: 6,
805                failure: Some(Failure {
806                    message: "out of retries".into(),
807                    ..Default::default()
808                }),
809                ..Default::default()
810            }),
811        ));
812
813        assert_eq!(n.group, GroupRef::Opened(5));
814        assert_eq!(n.role, Role::Closes);
815        assert_eq!(n.outcome, Outcome::Failed);
816        assert!(n.outcome.is_failure());
817        assert_eq!(n.failure.as_deref(), Some("out of retries"));
818    }
819
820    #[test]
821    fn a_timer_is_grouped_by_its_started_event_not_a_scheduled_one() {
822        // Timers are the one family that back-references `started_event_id`. Reaching for
823        // `scheduled_event_id` out of habit would leave every fired timer orphaned.
824        let started = normalize(event(
825            10,
826            EventType::TimerStarted,
827            Attributes::TimerStartedEventAttributes(TimerStartedEventAttributes {
828                timer_id: "sleep-1".into(),
829                workflow_task_completed_event_id: 9,
830                ..Default::default()
831            }),
832        ));
833        let fired = normalize(event(
834            11,
835            EventType::TimerFired,
836            Attributes::TimerFiredEventAttributes(TimerFiredEventAttributes {
837                timer_id: "sleep-1".into(),
838                started_event_id: 10,
839            }),
840        ));
841
842        assert_eq!(started.group, GroupRef::Opened(10));
843        assert_eq!(started.subject, "sleep-1");
844        assert_eq!(fired.group, started.group, "the fired timer must join it");
845        assert_eq!(fired.outcome, Outcome::Completed);
846    }
847
848    #[test]
849    fn the_workflow_start_event_belongs_to_the_workflow_group() {
850        let n = normalize(event(
851            1,
852            EventType::WorkflowExecutionStarted,
853            Attributes::WorkflowExecutionStartedEventAttributes(
854                WorkflowExecutionStartedEventAttributes {
855                    workflow_type: Some(
856                        temporalio_common::protos::temporal::api::common::v1::WorkflowType {
857                            name: "OrderWorkflow".into(),
858                        },
859                    ),
860                    attempt: 1,
861                    ..Default::default()
862                },
863            ),
864        ));
865
866        assert_eq!(n.group, GroupRef::Workflow);
867        assert_eq!(n.role, Role::Opens);
868        assert_eq!(n.subject, "OrderWorkflow");
869    }
870
871    #[test]
872    fn an_event_with_no_attributes_is_kept_rather_than_dropped() {
873        let mut e = HistoryEvent {
874            event_id: 99,
875            ..Default::default()
876        };
877        e.attributes = None;
878        let n = normalize(e);
879        assert_eq!(n.id, 99);
880        assert_eq!(n.group, GroupRef::Workflow);
881    }
882
883    fn json_payload(body: &str) -> ProtoPayload {
884        ProtoPayload {
885            metadata: [("encoding".to_string(), b"json/plain".to_vec())]
886                .into_iter()
887                .collect(),
888            data: body.as_bytes().to_vec(),
889            external_payloads: Vec::new(),
890        }
891    }
892
893    #[test]
894    fn an_activity_carries_its_input_payload() {
895        let n = normalize(event(
896            5,
897            EventType::ActivityTaskScheduled,
898            Attributes::ActivityTaskScheduledEventAttributes(
899                ActivityTaskScheduledEventAttributes {
900                    activity_type: Some(ActivityType {
901                        name: "ChargeCard".into(),
902                    }),
903                    input: Some(Payloads {
904                        payloads: vec![json_payload("100")],
905                    }),
906                    ..Default::default()
907                },
908            ),
909        ));
910
911        assert_eq!(n.payloads.len(), 1);
912        let (label, p) = &n.payloads[0];
913        assert_eq!(label, "input", "a lone argument is not indexed");
914        assert_eq!(p.encoding, "json/plain");
915        assert_eq!(
916            p.render(),
917            tmprl_core::payload::Rendered::Text("100".into())
918        );
919    }
920
921    #[test]
922    fn several_arguments_are_indexed() {
923        // An activity's third argument is not interchangeable with its first, so an
924        // unlabelled list would lose which is which.
925        let n = normalize(event(
926            5,
927            EventType::ActivityTaskScheduled,
928            Attributes::ActivityTaskScheduledEventAttributes(
929                ActivityTaskScheduledEventAttributes {
930                    input: Some(Payloads {
931                        payloads: vec![json_payload("1"), json_payload("\"two\"")],
932                    }),
933                    ..Default::default()
934                },
935            ),
936        ));
937
938        let labels: Vec<&str> = n.payloads.iter().map(|(l, _)| l.as_str()).collect();
939        assert_eq!(labels, ["input[0]", "input[1]"]);
940    }
941
942    #[test]
943    fn an_event_with_no_payloads_carries_none() {
944        let n = normalize(event(
945            6,
946            EventType::ActivityTaskStarted,
947            Attributes::ActivityTaskStartedEventAttributes(ActivityTaskStartedEventAttributes {
948                scheduled_event_id: 5,
949                ..Default::default()
950            }),
951        ));
952        assert!(n.payloads.is_empty());
953    }
954
955    #[test]
956    fn payload_metadata_is_decoded_from_bytes() {
957        // Metadata values are bytes on the wire; the encoding is what decides how the value
958        // is shown, so getting it out wrongly would make every payload opaque.
959        let mut raw = json_payload("{}");
960        raw.metadata.insert("type".to_string(), b"Keyword".to_vec());
961        let p = convert(raw);
962        assert_eq!(p.encoding, "json/plain");
963        assert_eq!(p.type_hint.as_deref(), Some("Keyword"));
964    }
965
966    #[test]
967    fn a_workflow_carries_its_own_input() {
968        // The first thing anyone opens a history to see. Reading only the activity events
969        // leaves the workflow's own arguments invisible.
970        let n = normalize(event(
971            1,
972            EventType::WorkflowExecutionStarted,
973            Attributes::WorkflowExecutionStartedEventAttributes(
974                WorkflowExecutionStartedEventAttributes {
975                    input: Some(Payloads {
976                        payloads: vec![json_payload(r#"{"orderId":7}"#)],
977                    }),
978                    ..Default::default()
979                },
980            ),
981        ));
982
983        let (label, p) = &n.payloads[0];
984        assert_eq!(label, "input");
985        assert_eq!(
986            p.render(),
987            tmprl_core::payload::Rendered::Text("{\n  \"orderId\": 7\n}".into())
988        );
989    }
990
991    #[test]
992    fn a_workflow_carries_its_own_result() {
993        let n = normalize(event(
994            11,
995            EventType::WorkflowExecutionCompleted,
996            Attributes::WorkflowExecutionCompletedEventAttributes(
997                WorkflowExecutionCompletedEventAttributes {
998                    result: Some(Payloads {
999                        payloads: vec![json_payload(r#""ok""#)],
1000                    }),
1001                    ..Default::default()
1002                },
1003            ),
1004        ));
1005
1006        assert_eq!(n.outcome, Outcome::Completed);
1007        let labels: Vec<&str> = n.payloads.iter().map(|(l, _)| l.as_str()).collect();
1008        assert_eq!(labels, ["result"]);
1009    }
1010
1011    #[test]
1012    fn a_signal_carries_its_argument() {
1013        let n = normalize(event(
1014            8,
1015            EventType::WorkflowExecutionSignaled,
1016            Attributes::WorkflowExecutionSignaledEventAttributes(
1017                WorkflowExecutionSignaledEventAttributes {
1018                    signal_name: "approve".into(),
1019                    input: Some(Payloads {
1020                        payloads: vec![json_payload("true")],
1021                    }),
1022                    ..Default::default()
1023                },
1024            ),
1025        ));
1026
1027        assert_eq!(n.subject, "approve");
1028        assert_eq!(n.payloads.len(), 1);
1029    }
1030
1031    #[test]
1032    fn a_marker_details_map_is_ordered_by_key() {
1033        // The map has no order on the wire, and a detail pane that reshuffles between
1034        // refreshes cannot be read.
1035        let details = HashMap::from([
1036            (
1037                "side-effect-id".to_string(),
1038                Payloads {
1039                    payloads: vec![json_payload("1")],
1040                },
1041            ),
1042            (
1043                "data".to_string(),
1044                Payloads {
1045                    payloads: vec![json_payload("2")],
1046                },
1047            ),
1048        ]);
1049        let n = normalize(event(
1050            9,
1051            EventType::MarkerRecorded,
1052            Attributes::MarkerRecordedEventAttributes(MarkerRecordedEventAttributes {
1053                marker_name: "SideEffect".into(),
1054                details,
1055                ..Default::default()
1056            }),
1057        ));
1058
1059        let labels: Vec<&str> = n.payloads.iter().map(|(l, _)| l.as_str()).collect();
1060        assert_eq!(labels, ["data", "side-effect-id"]);
1061    }
1062
1063    #[test]
1064    fn an_update_that_the_workflow_rejected_is_not_reported_as_completed() {
1065        // The event type says Completed either way; only the outcome oneof distinguishes a
1066        // successful update from one the workflow refused.
1067        let n = normalize(event(
1068            12,
1069            EventType::WorkflowExecutionUpdateCompleted,
1070            Attributes::WorkflowExecutionUpdateCompletedEventAttributes(
1071                WorkflowExecutionUpdateCompletedEventAttributes {
1072                    outcome: Some(UpdateOutcome {
1073                        value: Some(OutcomeValue::Failure(Failure {
1074                            message: "not allowed in this state".into(),
1075                            ..Default::default()
1076                        })),
1077                    }),
1078                    ..Default::default()
1079                },
1080            ),
1081        ));
1082
1083        assert_eq!(n.outcome, Outcome::Failed);
1084        assert_eq!(n.failure.as_deref(), Some("not allowed in this state"));
1085        assert!(n.payloads.is_empty());
1086    }
1087
1088    #[test]
1089    fn an_update_that_succeeded_carries_its_result() {
1090        let n = normalize(event(
1091            12,
1092            EventType::WorkflowExecutionUpdateCompleted,
1093            Attributes::WorkflowExecutionUpdateCompletedEventAttributes(
1094                WorkflowExecutionUpdateCompletedEventAttributes {
1095                    outcome: Some(UpdateOutcome {
1096                        value: Some(OutcomeValue::Success(Payloads {
1097                            payloads: vec![json_payload("42")],
1098                        })),
1099                    }),
1100                    ..Default::default()
1101                },
1102            ),
1103        ));
1104
1105        assert_eq!(n.outcome, Outcome::Completed);
1106        let labels: Vec<&str> = n.payloads.iter().map(|(l, _)| l.as_str()).collect();
1107        assert_eq!(labels, ["result"]);
1108    }
1109
1110    #[test]
1111    fn a_page_knows_whether_more_exist() {
1112        assert!(!HistoryPage::default().has_more());
1113        assert!(
1114            HistoryPage {
1115                next_page_token: vec![1],
1116                ..Default::default()
1117            }
1118            .has_more()
1119        );
1120    }
1121}
tmprl_client/ops/history.rs

tmprl_client/ops/
history.rs
