aion_store/visibility/record.rs
1//! The visibility row, the store contract, and the page it answers with.
2
3use std::collections::HashMap;
4
5use aion_core::{
6 OutstandingLease, PackageVersion, RunId, SearchAttributeValue, WorkflowId, WorkflowListRequest,
7 WorkflowStatus, WorkflowSummary,
8};
9use async_trait::async_trait;
10use chrono::{DateTime, Utc};
11use serde::{Deserialize, Serialize};
12
13use crate::StoreError;
14
15/// The visibility store: rows in, pages out.
16///
17/// ONE ROW PER WORKFLOW IDENTITY. A workflow that continues as new keeps one
18/// history and one row: the successor generation's upsert REPLACES the
19/// predecessor's row in place, so a workloop is a single row for its whole
20/// life and a list never shows generations. `run_id` stays on the row as
21/// data — the current generation — and the generation chain remains readable
22/// from history under describe.
23///
24/// `record_visibility` is the ONLY write; the engine's Recorder calls it after
25/// every durable append it projects (lifecycle events re-project the whole
26/// row, every other append advances `updated_at`). `list_workflows` answers
27/// the list contract from rows alone — a backend never reads history here.
28#[async_trait]
29pub trait VisibilityStore: Send + Sync + 'static {
30 /// Upserts THE row for `record.workflow_id`, replacing any previous
31 /// generation's row and moving it under its new sort keys atomically.
32 /// The replace is part of the write, not a separate pass: after this
33 /// returns, the workflow has exactly one row and it is this one.
34 ///
35 /// # Errors
36 ///
37 /// Returns backend or serialization errors encountered while writing.
38 async fn record_visibility(&self, record: VisibilityRecord) -> Result<(), StoreError>;
39
40 /// Reads the workflow's row, or `None` when it has no row yet.
41 ///
42 /// # Errors
43 ///
44 /// Returns backend or serialization errors encountered while reading.
45 async fn get_visibility(
46 &self,
47 workflow_id: &WorkflowId,
48 ) -> Result<Option<VisibilityRecord>, StoreError>;
49
50 /// Removes the row for `workflow_id` IF it belongs to `run_id`, and any
51 /// row a store written before the one-row collapse still holds for that
52 /// `(workflow_id, run_id)` pair. Returns whether anything was removed.
53 ///
54 /// The reconciler is the caller: it prunes every generation that is not
55 /// the current one, which is how a pre-collapse store converges to one
56 /// row per workflow at its next boot without a migration step. The run
57 /// guard makes the prune safe against racing a newer generation's upsert:
58 /// a remove naming a superseded run never deletes the current row.
59 ///
60 /// # Errors
61 ///
62 /// Returns backend or serialization errors encountered while removing.
63 async fn remove_visibility(
64 &self,
65 workflow_id: &WorkflowId,
66 run_id: &RunId,
67 ) -> Result<bool, StoreError>;
68
69 /// Answers one page of the list contract: `request.filter` applied
70 /// BEFORE the limit, ordered by `request.sort` with `workflow_id` as the
71 /// total-order tie-break, continuing after `request.cursor`, at most
72 /// `request.limit` rows, with the filtered total.
73 ///
74 /// # Errors
75 ///
76 /// Returns [`StoreError::InvalidQuery`] for a zero limit or a cursor that
77 /// was not minted under this exact `(namespace, filter, sort)`, and
78 /// backend or serialization errors encountered while reading.
79 async fn list_workflows(
80 &self,
81 request: &WorkflowListRequest,
82 ) -> Result<VisibilityPage, StoreError>;
83}
84
85/// One page of rows, before the server turns them into wire summaries.
86#[derive(Clone, Debug, PartialEq)]
87pub struct VisibilityPage {
88 /// The rows, in the requested order.
89 pub items: Vec<VisibilityRecord>,
90 /// The cursor naming the page's last row, present only when more rows
91 /// follow it.
92 pub next_cursor: Option<String>,
93 /// How many rows match the filter in total, cursor disregarded.
94 pub count: u64,
95}
96
97/// The complete row for one workflow execution.
98///
99/// Every field a list can sort or filter on is a first-class field here,
100/// projected by the engine when the row is written; `search_attributes`
101/// remains the full recorded attribute map for readers that need more.
102#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
103pub struct VisibilityRecord {
104 /// The namespace the run was placed in, from the `aion.namespace` start
105 /// attribute (`aion_core::DEFAULT_NAMESPACE` for a run recorded with no
106 /// placement).
107 pub namespace: String,
108 /// Logical workflow identifier.
109 pub workflow_id: WorkflowId,
110 /// Concrete run identifier for this workflow execution.
111 pub run_id: RunId,
112 /// Workflow type recorded when the execution started.
113 pub workflow_type: String,
114 /// Current projected status.
115 pub status: WorkflowStatus,
116 /// Recorded instant of the run's `WorkflowStarted`.
117 pub started_at: DateTime<Utc>,
118 /// Recorded instant of the run's latest event of any kind.
119 pub updated_at: DateTime<Utc>,
120 /// Recorded instant of the current lease's terminal event, if terminal.
121 pub ended_at: Option<DateTime<Utc>>,
122 /// The parent workflow, for a child execution.
123 pub parent: Option<WorkflowId>,
124 /// The operator-facing display name, from the `aion.display_name`
125 /// attribute (last write wins).
126 pub display_name: Option<String>,
127 /// The document kind from the `aion.kind` attribute (`"workloop"`), or
128 /// `None` for an ordinary workflow.
129 pub kind: Option<String>,
130 /// The step (activity type) that failed, only for a terminal-Failed run.
131 pub failed_step: Option<String>,
132 /// The terminal `WorkflowFailed` message, only for a terminal-Failed run.
133 pub failure_reason: Option<String>,
134 /// The full recorded search-attribute map.
135 pub search_attributes: HashMap<String, SearchAttributeValue>,
136 /// Every attempt a worker currently holds, earliest lease first — the
137 /// same fold as [`aion_core::outstanding_leases`] over the run's current
138 /// lease segment, maintained event by event by the engine's Recorder
139 /// through [`aion_core::apply_lease_transition`] (WA-010 R4). The
140 /// summary's `current_worker` is the last entry.
141 ///
142 /// Defaults to empty so a row written before the field existed decodes
143 /// as UNATTRIBUTED rather than failing to decode.
144 #[serde(default)]
145 pub outstanding_leases: Vec<OutstandingLease>,
146 /// The content hash of the package this run started under, from the run's
147 /// own `WorkflowStarted` (see [`WorkflowSummary::package_version`]).
148 ///
149 /// Defaults to `None` so a row written before the field existed decodes
150 /// as "not carried" rather than failing to decode — and is NEVER filled
151 /// in by looking up the type's current version.
152 #[serde(default)]
153 pub package_version: Option<PackageVersion>,
154}
155
156impl VisibilityRecord {
157 /// The wire summary this row projects to.
158 #[must_use]
159 pub fn summary(&self) -> WorkflowSummary {
160 WorkflowSummary {
161 workflow_id: self.workflow_id.clone(),
162 run_id: self.run_id.clone(),
163 workflow_type: self.workflow_type.clone(),
164 status: self.status,
165 started_at: self.started_at,
166 updated_at: self.updated_at,
167 ended_at: self.ended_at,
168 parent: self.parent.clone(),
169 failed_step: self.failed_step.clone(),
170 failure_reason: self.failure_reason.clone(),
171 display_name: self.display_name.clone(),
172 kind: self.kind.clone(),
173 current_worker: aion_core::current_worker(&self.outstanding_leases),
174 package_version: self.package_version.clone(),
175 }
176 }
177}
178
179impl From<VisibilityRecord> for WorkflowSummary {
180 fn from(record: VisibilityRecord) -> Self {
181 Self {
182 workflow_id: record.workflow_id,
183 run_id: record.run_id,
184 workflow_type: record.workflow_type,
185 status: record.status,
186 started_at: record.started_at,
187 updated_at: record.updated_at,
188 ended_at: record.ended_at,
189 parent: record.parent,
190 failed_step: record.failed_step,
191 failure_reason: record.failure_reason,
192 display_name: record.display_name,
193 kind: record.kind,
194 current_worker: aion_core::current_worker(&record.outstanding_leases),
195 package_version: record.package_version,
196 }
197 }
198}
199
200#[cfg(test)]
201mod tests {
202 use std::sync::Arc;
203
204 use aion_core::{PackageVersion, WorkflowSummary};
205
206 use super::{VisibilityRecord, VisibilityStore};
207
208 #[test]
209 fn visibility_store_is_object_safe() {
210 let _: Option<Arc<dyn VisibilityStore>> = None;
211 }
212
213 /// A row persisted before `package_version` (and `outstanding_leases`)
214 /// existed decodes as "not carried": the summary reports `None`, and
215 /// nothing fills it in from the type's current version. A row that
216 /// carries the hash reports exactly that hash.
217 #[test]
218 fn a_row_written_before_the_package_field_decodes_as_not_carried()
219 -> Result<(), Box<dyn std::error::Error>> {
220 let legacy = serde_json::json!({
221 "namespace": "default",
222 "workflow_id": "00000000-0000-0000-0000-000000000001",
223 "run_id": "00000000-0000-0000-0000-000000000002",
224 "workflow_type": "checkout",
225 "status": "Running",
226 "started_at": "2023-11-14T22:13:21Z",
227 "updated_at": "2023-11-14T22:13:21Z",
228 "ended_at": null,
229 "parent": null,
230 "display_name": null,
231 "kind": null,
232 "failed_step": null,
233 "failure_reason": null,
234 "search_attributes": {}
235 });
236 let record: VisibilityRecord = serde_json::from_value(legacy.clone())?;
237 assert_eq!(record.package_version, None);
238 assert_eq!(record.summary().package_version, None);
239
240 let mut carried = legacy;
241 carried["package_version"] = serde_json::Value::String("c".repeat(64));
242 let record: VisibilityRecord = serde_json::from_value(carried)?;
243 assert_eq!(
244 record.summary().package_version,
245 Some(PackageVersion::new("c".repeat(64)))
246 );
247 assert_eq!(
248 WorkflowSummary::from(record).package_version,
249 Some(PackageVersion::new("c".repeat(64)))
250 );
251 Ok(())
252 }
253}