tmprl_client/ops/
workflow.rs

1//! Listing and counting workflow executions.
2//!
3//! Two RPCs back the workflow table. `ListWorkflowExecutions` returns the rows, one page at
4//! a time; `CountWorkflowExecutions` with a `GROUP BY` returns the header tallies in a
5//! single cheap call, instead of the table having to page to exhaustion to know how many
6//! workflows are failing.
7//!
8//! Both take an explicit namespace rather than using the connection's own. A connection is
9//! bound to one namespace, but clones share a single HTTP/2 channel, so a fan-out across
10//! namespaces is one connection and N requests. See `list_workflows_across`.
11
12use temporalio_client::tonic::Request;
13use temporalio_common::protos::temporal::api::{
14    enums::v1::WorkflowExecutionStatus as ProtoStatus,
15    workflow::v1::WorkflowExecutionInfo,
16    workflowservice::v1::{CountWorkflowExecutionsRequest, ListWorkflowExecutionsRequest},
17};
18use tmprl_core::query::count_query;
19use tmprl_core::workflow::{StatusCounts, WorkflowRow, WorkflowStatus, merge_by_start_time};
20
21use super::OpError;
22use crate::Conn;
23
24/// Per-namespace continuation tokens. A namespace missing from the list has no more pages;
25/// an entry with an empty token starts from the beginning.
26pub type Continuation = Vec<(String, Vec<u8>)>;
27
28/// One page of the workflow table, plus the token that fetches the next.
29#[derive(Debug, Clone, Default)]
30pub struct WorkflowPage {
31    pub rows: Vec<WorkflowRow>,
32    /// Empty on the last page. This is opaque server state, never construct one.
33    pub next_page_token: Vec<u8>,
34}
35
36impl WorkflowPage {
37    /// Whether another page exists. Infinite scroll stops asking when this goes false.
38    pub fn has_more(&self) -> bool {
39        !self.next_page_token.is_empty()
40    }
41}
42
43impl Conn {
44    /// One page of executions matching `query`, newest first.
45    ///
46    /// This deliberately does *not* page to exhaustion the way `list_namespaces` does. A
47    /// namespace list is tens of rows; a workflow list is unbounded, and draining it would
48    /// hang the interface on any real cluster.
49    pub async fn list_workflows(
50        &self,
51        namespace: &str,
52        query: &str,
53        page_size: i32,
54        next_page_token: Vec<u8>,
55    ) -> Result<WorkflowPage, OpError> {
56        let resp = self
57            .wf()
58            .list_workflow_executions(Request::new(ListWorkflowExecutionsRequest {
59                namespace: namespace.to_string(),
60                page_size,
61                next_page_token,
62                query: query.to_string(),
63            }))
64            .await
65            .map_err(|s| OpError::rpc("ListWorkflowExecutions", s))?
66            .into_inner();
67
68        Ok(WorkflowPage {
69            rows: resp
70                .executions
71                .into_iter()
72                .map(|e| row_from(namespace, e))
73                .collect(),
74            next_page_token: resp.next_page_token,
75        })
76    }
77
78    /// The first page from several namespaces at once, merged newest-first.
79    ///
80    /// Returned alongside the rows is a continuation token *per namespace*, because the
81    /// namespaces exhaust at different points and one merged token cannot express that.
82    /// A namespace that is already finished simply does not appear in the returned list.
83    /// Feed that list to [`Conn::continue_workflows_across`] for the next page.
84    pub async fn list_workflows_across(
85        &self,
86        namespaces: &[String],
87        query: &str,
88        page_size: i32,
89    ) -> Result<(Vec<WorkflowRow>, Continuation), OpError> {
90        let starting: Continuation = namespaces
91            .iter()
92            .map(|ns| (ns.clone(), Vec::new()))
93            .collect();
94        self.fetch_pages(&starting, query, page_size).await
95    }
96
97    /// The next page, asking *only* the namespaces that still have one.
98    ///
99    /// Taking the token list rather than the namespace list is the point. Re-deriving the
100    /// namespaces from the original scope would hand an exhausted namespace an empty token,
101    /// which the server reads as "start from the beginning", so it would return page one
102    /// again, along with a fresh token, and that namespace would never finish.
103    pub async fn continue_workflows_across(
104        &self,
105        tokens: &Continuation,
106        query: &str,
107        page_size: i32,
108    ) -> Result<(Vec<WorkflowRow>, Continuation), OpError> {
109        self.fetch_pages(tokens, query, page_size).await
110    }
111
112    /// One request per namespace, in flight together. They share the connection's channel,
113    /// so this costs N streams rather than N connections.
114    async fn fetch_pages(
115        &self,
116        tokens: &Continuation,
117        query: &str,
118        page_size: i32,
119    ) -> Result<(Vec<WorkflowRow>, Continuation), OpError> {
120        let pages = futures_util::future::try_join_all(tokens.iter().map(|(ns, token)| {
121            let token = token.clone();
122            async move {
123                self.list_workflows(ns, query, page_size, token)
124                    .await
125                    .map(|p| (ns.clone(), p))
126            }
127        }))
128        .await?;
129
130        let mut next = Continuation::new();
131        let mut all = Vec::new();
132        for (ns, page) in pages {
133            if page.has_more() {
134                next.push((ns, page.next_page_token));
135            }
136            all.push(page.rows);
137        }
138        Ok((merge_by_start_time(all), next))
139    }
140
141    /// Per-status counts for the list header.
142    ///
143    /// One `GROUP BY` call rather than one call per status. The grouped counts are
144    /// approximate by design on large clusters (Temporal says so) which is why the total
145    /// is taken from the response rather than summed from the groups.
146    pub async fn count_workflows_by_status(
147        &self,
148        namespace: &str,
149        query: &str,
150    ) -> Result<StatusCounts, OpError> {
151        let resp = self
152            .wf()
153            .count_workflow_executions(Request::new(CountWorkflowExecutionsRequest {
154                namespace: namespace.to_string(),
155                query: count_query(query),
156            }))
157            .await
158            .map_err(|s| OpError::rpc("CountWorkflowExecutions", s))?
159            .into_inner();
160
161        let counts = resp.groups.into_iter().filter_map(|g| {
162            let status = g.group_values.first().and_then(status_from_payload)?;
163            Some((status, g.count))
164        });
165        Ok(StatusCounts::new(resp.count, counts))
166    }
167}
168
169impl Conn {
170    /// Header counts summed over a fan-out.
171    ///
172    /// A header that counted only the first of several namespaces would be quietly wrong,
173    /// which is worse than having no header at all.
174    pub async fn count_workflows_across(
175        &self,
176        namespaces: &[String],
177        query: &str,
178    ) -> Result<StatusCounts, OpError> {
179        let per_ns = futures_util::future::try_join_all(
180            namespaces
181                .iter()
182                .map(|ns| self.count_workflows_by_status(ns, query)),
183        )
184        .await?;
185
186        let mut total = 0;
187        let mut summed: Vec<(WorkflowStatus, i64)> = Vec::new();
188        for counts in &per_ns {
189            total += counts.total;
190            for (status, n) in counts.iter() {
191                match summed.iter_mut().find(|(s, _)| *s == status) {
192                    Some((_, acc)) => *acc += n,
193                    None => summed.push((status, n)),
194                }
195            }
196        }
197        Ok(StatusCounts::new(total, summed))
198    }
199}
200
201/// Map the protobuf row onto the domain row.
202fn row_from(namespace: &str, e: WorkflowExecutionInfo) -> WorkflowRow {
203    // `status()` borrows `e`, so resolve it before the string fields are moved out.
204    let status = status_from_proto(e.status());
205    let (workflow_id, run_id) = e
206        .execution
207        .map(|x| (x.workflow_id, x.run_id))
208        .unwrap_or_default();
209
210    WorkflowRow {
211        namespace: namespace.to_string(),
212        workflow_id,
213        run_id,
214        workflow_type: e.r#type.map(|t| t.name).unwrap_or_default(),
215        task_queue: e.task_queue,
216        status,
217        start_time: e.start_time.map(epoch_millis),
218        close_time: e.close_time.map(epoch_millis),
219        history_length: e.history_length,
220    }
221}
222
223/// Exhaustive on purpose. When Temporal adds an execution status this stops compiling,
224/// which is the moment we want to hear about it, a `_` arm would render the new status as
225/// `Unspecified` and nobody would notice for a release or two.
226fn status_from_proto(s: ProtoStatus) -> WorkflowStatus {
227    match s {
228        ProtoStatus::Unspecified => WorkflowStatus::Unspecified,
229        ProtoStatus::Running => WorkflowStatus::Running,
230        ProtoStatus::Completed => WorkflowStatus::Completed,
231        ProtoStatus::Failed => WorkflowStatus::Failed,
232        ProtoStatus::Canceled => WorkflowStatus::Canceled,
233        ProtoStatus::Terminated => WorkflowStatus::Terminated,
234        ProtoStatus::ContinuedAsNew => WorkflowStatus::ContinuedAsNew,
235        ProtoStatus::TimedOut => WorkflowStatus::TimedOut,
236        ProtoStatus::Paused => WorkflowStatus::Paused,
237    }
238}
239
240/// A `GROUP BY ExecutionStatus` group value is a `json/plain` Keyword payload whose data is
241/// the quoted status name, e.g. `"Running"`. Parsing tolerates the quotes.
242fn status_from_payload(
243    p: &temporalio_common::protos::temporal::api::common::v1::Payload,
244) -> Option<WorkflowStatus> {
245    WorkflowStatus::parse(std::str::from_utf8(&p.data).ok()?)
246}
247
248fn epoch_millis(t: prost_wkt_types::Timestamp) -> i64 {
249    t.seconds * 1000 + i64::from(t.nanos) / 1_000_000
250}
251
252#[cfg(test)]
253mod tests {
254    use super::*;
255    use temporalio_common::protos::temporal::api::common::v1::{
256        Payload, WorkflowExecution, WorkflowType,
257    };
258    use tmprl_core::workflow::WorkflowStatus;
259
260    fn payload(body: &str) -> Payload {
261        Payload {
262            metadata: [
263                ("encoding".to_string(), b"json/plain".to_vec()),
264                ("type".to_string(), b"Keyword".to_vec()),
265            ]
266            .into_iter()
267            .collect(),
268            data: body.as_bytes().to_vec(),
269            external_payloads: Vec::new(),
270        }
271    }
272
273    #[test]
274    fn every_proto_status_maps_to_a_domain_status() {
275        // Pairwise distinct: a copy-paste slip in the match would collapse two statuses.
276        let all = [
277            ProtoStatus::Unspecified,
278            ProtoStatus::Running,
279            ProtoStatus::Completed,
280            ProtoStatus::Failed,
281            ProtoStatus::Canceled,
282            ProtoStatus::Terminated,
283            ProtoStatus::ContinuedAsNew,
284            ProtoStatus::TimedOut,
285            ProtoStatus::Paused,
286        ];
287        let mut mapped: Vec<WorkflowStatus> = all.iter().copied().map(status_from_proto).collect();
288        mapped.sort_unstable();
289        mapped.dedup();
290        assert_eq!(mapped.len(), all.len(), "two proto statuses collapsed");
291    }
292
293    #[test]
294    fn group_payloads_decode_to_a_status() {
295        // The exact bytes a dev server returns for `GROUP BY ExecutionStatus`.
296        assert_eq!(
297            status_from_payload(&payload("\"Running\"")),
298            Some(WorkflowStatus::Running)
299        );
300        assert_eq!(
301            status_from_payload(&payload("\"ContinuedAsNew\"")),
302            Some(WorkflowStatus::ContinuedAsNew)
303        );
304        assert_eq!(status_from_payload(&payload("\"Nonsense\"")), None);
305    }
306
307    #[test]
308    fn timestamps_convert_to_epoch_millis() {
309        let t = prost_wkt_types::Timestamp {
310            seconds: 1_700_000_000,
311            nanos: 500_000_000,
312        };
313        assert_eq!(epoch_millis(t), 1_700_000_000_500);
314    }
315
316    #[test]
317    fn a_row_without_an_execution_still_maps() {
318        // Defensive: every field of WorkflowExecutionInfo is optional on the wire, and a
319        // panic here would take down the whole table for one malformed row.
320        let row = row_from("ns", WorkflowExecutionInfo::default());
321        assert_eq!(row.namespace, "ns");
322        assert!(row.workflow_id.is_empty() && row.run_id.is_empty());
323        assert_eq!(row.status, WorkflowStatus::Unspecified);
324        assert_eq!(row.start_time, None);
325    }
326
327    #[test]
328    fn a_populated_row_carries_its_namespace() {
329        let info = WorkflowExecutionInfo {
330            execution: Some(WorkflowExecution {
331                workflow_id: "wf-1".into(),
332                run_id: "run-1".into(),
333            }),
334            r#type: Some(WorkflowType {
335                name: "Greeter".into(),
336            }),
337            task_queue: "tq".into(),
338            status: ProtoStatus::Completed as i32,
339            history_length: 12,
340            start_time: Some(prost_wkt_types::Timestamp {
341                seconds: 100,
342                nanos: 0,
343            }),
344            ..Default::default()
345        };
346        let row = row_from("payments", info);
347        assert_eq!(row.namespace, "payments");
348        assert_eq!(row.workflow_id, "wf-1");
349        assert_eq!(row.workflow_type, "Greeter");
350        assert_eq!(row.status, WorkflowStatus::Completed);
351        assert_eq!(row.start_time, Some(100_000));
352        assert_eq!(row.history_length, 12);
353    }
354
355    #[test]
356    fn a_page_knows_whether_more_exist() {
357        assert!(!WorkflowPage::default().has_more());
358        assert!(
359            WorkflowPage {
360                next_page_token: vec![1],
361                ..Default::default()
362            }
363            .has_more()
364        );
365    }
366}
tmprl_client/ops/workflow.rs

tmprl_client/ops/
workflow.rs
