ironflow_api/entities/
run.rs

1//! Run-related DTOs and query parameters.
2
3use std::collections::HashMap;
4
5use chrono::{DateTime, Utc};
6use ironflow_store::models::{ConcurrencyLimit, ProviderKind, Run, RunStatus, TriggerKind};
7use rust_decimal::Decimal;
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10use uuid::Uuid;
11
12use super::{CreatedBy, StepResponse};
13
14/// Run response DTO — public API representation of a run.
15///
16/// Maps from the internal [`Run`] model, exposing only necessary fields.
17///
18/// # Examples
19///
20/// ```
21/// use ironflow_store::models::{Run, RunStatus, TriggerKind};
22/// use ironflow_api::entities::RunResponse;
23/// ```
24#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
25#[derive(Debug, Serialize, Deserialize)]
26pub struct RunResponse {
27    /// Unique run identifier.
28    pub id: Uuid,
29    /// Workflow name.
30    pub workflow_name: String,
31    /// Current status.
32    pub status: RunStatus,
33    /// How the run was triggered.
34    pub trigger: TriggerKind,
35    /// Optional error message.
36    pub error: Option<String>,
37    /// Number of times retried.
38    pub retry_count: u32,
39    /// Maximum allowed retries.
40    pub max_retries: u32,
41    /// Number of times the run was requeued after its worker lost its lease. Separate from `retry_count`: it does not consume `max_retries`.
42    pub lease_recoveries: u32,
43    /// Aggregated cost in USD.
44    #[cfg_attr(feature = "openapi", schema(value_type = f64))]
45    pub cost_usd: Decimal,
46    /// Total duration in milliseconds.
47    pub duration_ms: u64,
48    /// When created.
49    pub created_at: DateTime<Utc>,
50    /// When last updated.
51    pub updated_at: DateTime<Utc>,
52    /// When execution started.
53    pub started_at: Option<DateTime<Utc>>,
54    /// When execution completed.
55    pub completed_at: Option<DateTime<Utc>>,
56    /// Version of the handler that created this run.
57    pub handler_version: Option<String>,
58    /// User-defined key-value labels.
59    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
60    pub labels: HashMap<String, String>,
61    /// Scheduled execution time. `None` means the run executed immediately.
62    #[serde(default, skip_serializing_if = "Option::is_none")]
63    pub scheduled_at: Option<DateTime<Utc>>,
64    /// Provider kind the run is waiting capacity for, set only while the
65    /// run is `sleeping` because every targeted Provider Account is rate
66    /// limited. `scheduled_at` then holds when the run wakes.
67    #[serde(default, skip_serializing_if = "Option::is_none")]
68    #[cfg_attr(feature = "openapi", schema(value_type = Option<String>))]
69    pub capacity_wait_kind: Option<ProviderKind>,
70    /// Who triggered the run. Always present.
71    pub created_by: CreatedBy,
72    /// Idempotency key that produced this run, when one was supplied.
73    #[serde(default, skip_serializing_if = "Option::is_none")]
74    pub idempotency_key: Option<String>,
75    /// Exclusivity key held by this run until it reaches a terminal state,
76    /// when one was supplied.
77    #[serde(default, skip_serializing_if = "Option::is_none")]
78    pub concurrency_key: Option<String>,
79    /// Queue priority, from -100 to 100. Workers pick the pending run with
80    /// the highest priority first, then the oldest among equal priorities.
81    /// `0` is the default.
82    #[serde(default)]
83    pub priority: i16,
84    /// Concurrency groups the run belongs to, with the limit it was created
85    /// with. Omitted when the run belongs to no group.
86    #[serde(default, skip_serializing_if = "Vec::is_empty")]
87    pub concurrency_limits: Vec<ConcurrencyLimit>,
88    /// Cumulative cost cap for this run, in USD. `None` means no cap.
89    #[cfg_attr(feature = "openapi", schema(value_type = Option<f64>))]
90    #[serde(default, skip_serializing_if = "Option::is_none")]
91    pub max_cost_usd: Option<Decimal>,
92    /// Tags a worker must carry to take this run, sorted. Empty when any
93    /// worker can take it.
94    #[serde(default)]
95    pub worker_tags: Vec<String>,
96    /// Typed output the handler set with `WorkflowContext::set_output`.
97    ///
98    /// Written when the run ends (completed, warning, failed or cancelled).
99    /// Omitted when the handler set no output.
100    #[serde(default, skip_serializing_if = "Option::is_none")]
101    pub output: Option<Value>,
102    /// State a `paused` run returns to when it is resumed. Omitted in every
103    /// other state.
104    #[serde(default, skip_serializing_if = "Option::is_none")]
105    pub resume_status: Option<RunStatus>,
106}
107
108impl From<Run> for RunResponse {
109    fn from(run: Run) -> Self {
110        let created_by = CreatedBy::from(&run);
111        RunResponse {
112            id: run.id,
113            workflow_name: run.workflow_name,
114            status: run.status.state,
115            trigger: run.trigger,
116            error: run.error,
117            retry_count: run.retry_count,
118            max_retries: run.max_retries,
119            lease_recoveries: run.lease_recoveries,
120            cost_usd: run.cost_usd,
121            duration_ms: run.duration_ms,
122            created_at: run.created_at,
123            updated_at: run.updated_at,
124            started_at: run.started_at,
125            completed_at: run.completed_at,
126            handler_version: run.handler_version,
127            labels: run.labels,
128            scheduled_at: run.scheduled_at,
129            capacity_wait_kind: run.capacity_wait_kind,
130            created_by,
131            idempotency_key: run.idempotency_key,
132            concurrency_key: run.concurrency_key,
133            priority: run.priority,
134            concurrency_limits: run.concurrency_limits,
135            max_cost_usd: run.max_cost_usd,
136            worker_tags: run.worker_tags,
137            output: run.output,
138            resume_status: run.resume_status,
139        }
140    }
141}
142
143/// Run detail response — includes steps and payload.
144#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
145#[derive(Debug, Serialize)]
146pub struct RunDetailResponse {
147    /// The run.
148    pub run: RunResponse,
149    /// Associated steps, ordered by position.
150    pub steps: Vec<StepResponse>,
151    /// Input payload that triggered this run.
152    pub payload: serde_json::Value,
153    /// Sub-workflow runs below this run, at any depth, that are not finished:
154    /// cancelling the run cancels them too.
155    pub active_descendant_count: u64,
156    /// Workers seen recently that could take the run. Only filled while the
157    /// run waits in the queue (`pending` or `retrying`), so a run no worker
158    /// can take is visible instead of waiting in silence.
159    #[serde(default, skip_serializing_if = "Option::is_none")]
160    pub worker_routing: Option<WorkerRouting>,
161}
162
163/// Workers the API saw recently, as seen from one queued run.
164///
165/// Counted from the pick requests received in the last few minutes by this
166/// API process: with several replicas, each one only counts the workers that
167/// polled it.
168///
169/// # Examples
170///
171/// ```
172/// use ironflow_api::entities::WorkerRouting;
173///
174/// let routing = WorkerRouting {
175///     seen_workers: 2,
176///     eligible_workers: 0,
177/// };
178/// assert!(routing.eligible_workers < routing.seen_workers);
179/// ```
180#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
181#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
182pub struct WorkerRouting {
183    /// Workers that asked for a run recently.
184    pub seen_workers: u32,
185    /// Among them, the workers that registered the run's workflow and carry
186    /// every tag it requires.
187    pub eligible_workers: u32,
188}
189
190/// Response of `POST /api/v1/runs/:id/cancel`: the cancelled run, with the
191/// sub-workflow runs cancelled along with it.
192///
193/// The run's fields stay at the top level, as before the descendants were
194/// listed, so existing clients read the same document.
195#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
196#[derive(Debug, Serialize)]
197pub struct CancelRunResponse {
198    /// The cancelled run.
199    #[serde(flatten)]
200    pub run: RunResponse,
201    /// Sub-workflow runs below it that this request cancelled, oldest first.
202    /// Empty when none was still active.
203    pub cancelled_descendants: Vec<Uuid>,
204}
205
206/// Response of `POST /api/v1/runs/:id/pause`: the paused run, with the
207/// sub-workflow runs paused along with it.
208#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
209#[derive(Debug, Serialize)]
210pub struct PauseRunResponse {
211    /// The paused run. Its `resume_status` is the state it returns to.
212    #[serde(flatten)]
213    pub run: RunResponse,
214    /// Sub-workflow runs below it that this request paused, oldest first.
215    pub paused_descendants: Vec<Uuid>,
216}
217
218/// Response of `POST /api/v1/runs/:id/resume`: the resumed run, with the
219/// sub-workflow runs resumed along with it.
220#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
221#[derive(Debug, Serialize)]
222pub struct ResumeRunResponse {
223    /// The resumed run, in the state it was paused from.
224    #[serde(flatten)]
225    pub run: RunResponse,
226    /// Sub-workflow runs below it that this request resumed, oldest first.
227    pub resumed_descendants: Vec<Uuid>,
228}
229
230/// Query parameters for listing runs.
231#[cfg_attr(feature = "openapi", derive(utoipa::IntoParams, utoipa::ToSchema))]
232#[derive(Debug, Deserialize)]
233pub struct ListRunsQuery {
234    /// Filter by workflow name.
235    pub workflow: Option<String>,
236    /// Filter by run status.
237    pub status: Option<RunStatus>,
238    /// Filter by step presence (only applies to completed/cancelled runs).
239    /// Non-terminal runs (pending, running, etc.) are always included.
240    /// When `true`, only return completed/cancelled runs that have steps.
241    /// When `false`, only return completed/cancelled runs without steps.
242    pub has_steps: Option<bool>,
243    /// Filter by labels. Comma-separated `key:value` pairs.
244    pub label: Option<String>,
245    /// Filter by author: the user ID that triggered the run.
246    ///
247    /// Also matches runs triggered by one of that user's API keys.
248    pub created_by: Option<Uuid>,
249    /// Filter by concurrency group: only runs that belong to this group.
250    pub concurrency_group: Option<String>,
251    /// Filter by priority: only runs with exactly this priority.
252    pub priority: Option<i16>,
253    /// Page number (1-based).
254    pub page: Option<u32>,
255    /// Items per page.
256    pub per_page: Option<u32>,
257}
258
259impl ListRunsQuery {
260    /// Parse the comma-separated `label` param into a `HashMap`.
261    pub fn parse_labels(&self) -> Option<HashMap<String, String>> {
262        parse_label_param(&self.label)
263    }
264}
265
266/// Parse a comma-separated `key:value` label query param into a `HashMap`.
267///
268/// Entries without a `:` are ignored. Returns `None` when the param is
269/// absent or contains no valid entry.
270pub(crate) fn parse_label_param(raw: &Option<String>) -> Option<HashMap<String, String>> {
271    raw.as_ref().and_then(|raw| {
272        let mut map = HashMap::new();
273        for entry in raw.split(',') {
274            let entry = entry.trim();
275            if let Some((k, v)) = entry.split_once(':') {
276                map.insert(k.to_string(), v.to_string());
277            }
278        }
279        if map.is_empty() { None } else { Some(map) }
280    })
281}
282
283#[cfg(test)]
284mod tests {
285    use ironflow_store::memory::InMemoryStore;
286    use ironflow_store::models::NewRun;
287    use ironflow_store::store::RunStore;
288    use serde_json::{json, to_value};
289
290    use super::*;
291
292    async fn run_with_lease_recoveries(lease_recoveries: u32) -> Run {
293        let store = InMemoryStore::new();
294        let mut run = store
295            .create_run(NewRun {
296                created_by: None,
297                workflow_name: "deploy".to_string(),
298                trigger: TriggerKind::Manual,
299                payload: json!({}),
300                max_retries: 1,
301                handler_version: None,
302                labels: HashMap::new(),
303                scheduled_at: None,
304                idempotency_key: None,
305                concurrency_key: None,
306                priority: 0,
307                concurrency_limits: Vec::new(),
308                max_cost_usd: None,
309                worker_tags: Vec::new(),
310            })
311            .await
312            .expect("create run")
313            .into_run();
314        run.lease_recoveries = lease_recoveries;
315        run
316    }
317
318    #[tokio::test]
319    async fn run_response_exposes_the_lease_recoveries() {
320        let response = RunResponse::from(run_with_lease_recoveries(2).await);
321
322        assert_eq!(response.lease_recoveries, 2);
323        let value = to_value(&response).expect("serialize");
324        assert_eq!(value["lease_recoveries"], 2);
325    }
326
327    #[tokio::test]
328    async fn run_response_reports_zero_lease_recoveries_for_a_fresh_run() {
329        let response = RunResponse::from(run_with_lease_recoveries(0).await);
330
331        assert_eq!(response.lease_recoveries, 0);
332        let value = to_value(&response).expect("serialize");
333        assert_eq!(value["lease_recoveries"], 0);
334    }
335}
ironflow_api/entities/run.rs

ironflow_api/entities/
run.rs
