harn_vm/
waitpoints.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::sync::Arc;
3
4use serde::{Deserialize, Serialize};
5
6use crate::event_log::{
7    sanitize_topic_component, AnyEventLog, EventLog, LogError, LogEvent, Topic,
8};
9
10pub const WAITPOINT_STATE_TOPIC_PREFIX: &str = "waitpoint.state.";
11pub const WAITPOINT_WAITS_TOPIC: &str = "waitpoint.waits";
12
13#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
14#[serde(rename_all = "snake_case")]
15pub enum WaitpointStatus {
16    #[default]
17    Open,
18    Completed,
19    Cancelled,
20}
21
22impl WaitpointStatus {
23    pub fn as_str(self) -> &'static str {
24        match self {
25            Self::Open => "open",
26            Self::Completed => "completed",
27            Self::Cancelled => "cancelled",
28        }
29    }
30}
31
32#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(rename_all = "snake_case")]
34pub enum WaitpointWaitStatus {
35    Completed,
36    Cancelled,
37    TimedOut,
38    Interrupted,
39}
40
41impl WaitpointWaitStatus {
42    pub fn as_str(self) -> &'static str {
43        match self {
44            Self::Completed => "completed",
45            Self::Cancelled => "cancelled",
46            Self::TimedOut => "timed_out",
47            Self::Interrupted => "interrupted",
48        }
49    }
50
51    fn event_kind(self) -> &'static str {
52        match self {
53            Self::Completed => "waitpoint_wait_completed",
54            Self::Cancelled => "waitpoint_wait_cancelled",
55            Self::TimedOut => "waitpoint_wait_timed_out",
56            Self::Interrupted => "waitpoint_wait_interrupted",
57        }
58    }
59}
60
61#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
62pub struct WaitpointRecord {
63    pub id: String,
64    pub status: WaitpointStatus,
65    pub created_at: String,
66    pub created_by: Option<String>,
67    pub completed_at: Option<String>,
68    pub completed_by: Option<String>,
69    pub cancelled_at: Option<String>,
70    pub cancelled_by: Option<String>,
71    pub reason: Option<String>,
72    #[serde(default)]
73    pub metadata: BTreeMap<String, serde_json::Value>,
74}
75
76impl WaitpointRecord {
77    pub fn open(
78        id: impl Into<String>,
79        created_by: Option<String>,
80        metadata: BTreeMap<String, serde_json::Value>,
81    ) -> Self {
82        Self {
83            id: id.into(),
84            status: WaitpointStatus::Open,
85            created_at: now_rfc3339(),
86            created_by,
87            completed_at: None,
88            completed_by: None,
89            cancelled_at: None,
90            cancelled_by: None,
91            reason: None,
92            metadata,
93        }
94    }
95
96    pub fn is_terminal(&self) -> bool {
97        matches!(
98            self.status,
99            WaitpointStatus::Completed | WaitpointStatus::Cancelled
100        )
101    }
102}
103
104#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
105pub struct WaitpointWaitStartRecord {
106    pub wait_id: String,
107    pub waitpoint_ids: Vec<String>,
108    pub started_at: String,
109    pub trace_id: Option<String>,
110    pub replay_of_event_id: Option<String>,
111}
112
113#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
114pub struct WaitpointWaitRecord {
115    pub wait_id: String,
116    pub waitpoint_ids: Vec<String>,
117    pub status: WaitpointWaitStatus,
118    pub started_at: String,
119    pub resolved_at: String,
120    pub waitpoints: Vec<WaitpointRecord>,
121    pub cancelled_waitpoint_id: Option<String>,
122    pub trace_id: Option<String>,
123    pub replay_of_event_id: Option<String>,
124    pub reason: Option<String>,
125}
126
127#[derive(Clone, Debug, PartialEq, Eq)]
128pub enum WaitpointResolution {
129    Pending,
130    Completed,
131    Cancelled { waitpoint_id: String },
132}
133
134pub fn dedupe_waitpoint_ids(ids: &[String]) -> Vec<String> {
135    let mut seen = BTreeSet::new();
136    let mut out = Vec::new();
137    for id in ids {
138        let trimmed = id.trim();
139        if trimmed.is_empty() {
140            continue;
141        }
142        if seen.insert(trimmed.to_string()) {
143            out.push(trimmed.to_string());
144        }
145    }
146    out
147}
148
149pub fn waitpoint_topic(id: &str) -> Result<Topic, LogError> {
150    Topic::new(format!(
151        "{WAITPOINT_STATE_TOPIC_PREFIX}{}",
152        sanitize_topic_component(id)
153    ))
154}
155
156pub fn waits_topic() -> Result<Topic, LogError> {
157    Topic::new(WAITPOINT_WAITS_TOPIC)
158}
159
160pub async fn load_waitpoint(
161    log: &Arc<AnyEventLog>,
162    id: &str,
163) -> Result<Option<WaitpointRecord>, LogError> {
164    let events = log
165        .read_range(&waitpoint_topic(id)?, None, usize::MAX)
166        .await?;
167    let mut latest = None;
168    for (_, event) in events {
169        if !matches!(
170            event.kind.as_str(),
171            "waitpoint_created" | "waitpoint_completed" | "waitpoint_cancelled"
172        ) {
173            continue;
174        }
175        let Ok(record) = serde_json::from_value::<WaitpointRecord>(event.payload) else {
176            continue;
177        };
178        latest = Some(record);
179    }
180    Ok(latest)
181}
182
183pub async fn load_waitpoints(
184    log: &Arc<AnyEventLog>,
185    ids: &[String],
186) -> Result<Vec<WaitpointRecord>, LogError> {
187    let mut out = Vec::new();
188    for id in dedupe_waitpoint_ids(ids) {
189        if let Some(record) = load_waitpoint(log, &id).await? {
190            out.push(record);
191        }
192    }
193    Ok(out)
194}
195
196pub fn resolve_waitpoints(ids: &[String], waitpoints: &[WaitpointRecord]) -> WaitpointResolution {
197    let mut by_id = BTreeMap::new();
198    for waitpoint in waitpoints {
199        by_id.insert(waitpoint.id.as_str(), waitpoint);
200    }
201    let ids = dedupe_waitpoint_ids(ids);
202    if ids.is_empty() {
203        return WaitpointResolution::Pending;
204    }
205
206    let mut all_completed = true;
207    for id in ids {
208        let Some(waitpoint) = by_id.get(id.as_str()) else {
209            all_completed = false;
210            continue;
211        };
212        match waitpoint.status {
213            WaitpointStatus::Completed => {}
214            WaitpointStatus::Cancelled => {
215                return WaitpointResolution::Cancelled {
216                    waitpoint_id: waitpoint.id.clone(),
217                };
218            }
219            WaitpointStatus::Open => {
220                all_completed = false;
221            }
222        }
223    }
224
225    if all_completed {
226        WaitpointResolution::Completed
227    } else {
228        WaitpointResolution::Pending
229    }
230}
231
232pub async fn create_waitpoint(
233    log: &Arc<AnyEventLog>,
234    id: &str,
235    created_by: Option<String>,
236    metadata: BTreeMap<String, serde_json::Value>,
237) -> Result<WaitpointRecord, LogError> {
238    if let Some(existing) = load_waitpoint(log, id).await? {
239        return Ok(existing);
240    }
241    let record = WaitpointRecord::open(id, created_by, metadata);
242    append_waitpoint_state(log, "waitpoint_created", &record).await?;
243    Ok(record)
244}
245
246pub async fn complete_waitpoint(
247    log: &Arc<AnyEventLog>,
248    id: &str,
249    completed_by: Option<String>,
250) -> Result<WaitpointRecord, LogError> {
251    let existing = load_waitpoint(log, id).await?;
252    if let Some(existing) = existing.as_ref() {
253        if existing.is_terminal() {
254            return Ok(existing.clone());
255        }
256    }
257
258    let now = now_rfc3339();
259    let mut record = existing.unwrap_or_else(|| WaitpointRecord {
260        id: id.to_string(),
261        status: WaitpointStatus::Open,
262        created_at: now.clone(),
263        created_by: completed_by.clone(),
264        completed_at: None,
265        completed_by: None,
266        cancelled_at: None,
267        cancelled_by: None,
268        reason: None,
269        metadata: BTreeMap::new(),
270    });
271    record.status = WaitpointStatus::Completed;
272    record.completed_at = Some(now);
273    record.completed_by = completed_by;
274    record.cancelled_at = None;
275    record.cancelled_by = None;
276    record.reason = None;
277    append_waitpoint_state(log, "waitpoint_completed", &record).await?;
278    Ok(record)
279}
280
281pub async fn cancel_waitpoint(
282    log: &Arc<AnyEventLog>,
283    id: &str,
284    cancelled_by: Option<String>,
285    reason: Option<String>,
286) -> Result<WaitpointRecord, LogError> {
287    let existing = load_waitpoint(log, id).await?;
288    if let Some(existing) = existing.as_ref() {
289        if existing.is_terminal() {
290            return Ok(existing.clone());
291        }
292    }
293
294    let now = now_rfc3339();
295    let mut record = existing.unwrap_or_else(|| WaitpointRecord {
296        id: id.to_string(),
297        status: WaitpointStatus::Open,
298        created_at: now.clone(),
299        created_by: cancelled_by.clone(),
300        completed_at: None,
301        completed_by: None,
302        cancelled_at: None,
303        cancelled_by: None,
304        reason: None,
305        metadata: BTreeMap::new(),
306    });
307    record.status = WaitpointStatus::Cancelled;
308    record.completed_at = None;
309    record.completed_by = None;
310    record.cancelled_at = Some(now);
311    record.cancelled_by = cancelled_by;
312    record.reason = reason;
313    append_waitpoint_state(log, "waitpoint_cancelled", &record).await?;
314    Ok(record)
315}
316
317pub async fn append_wait_started(
318    log: &Arc<AnyEventLog>,
319    record: &WaitpointWaitStartRecord,
320) -> Result<(), LogError> {
321    log.append(
322        &waits_topic()?,
323        LogEvent::new(
324            "waitpoint_wait_started",
325            serde_json::to_value(record).map_err(|error| {
326                LogError::Serde(format!("waitpoint wait encode error: {error}"))
327            })?,
328        )
329        .with_headers(wait_headers(&record.wait_id, &record.waitpoint_ids)),
330    )
331    .await
332    .map(|_| ())
333}
334
335pub async fn append_wait_terminal(
336    log: &Arc<AnyEventLog>,
337    record: &WaitpointWaitRecord,
338) -> Result<(), LogError> {
339    log.append(
340        &waits_topic()?,
341        LogEvent::new(
342            record.status.event_kind(),
343            serde_json::to_value(record).map_err(|error| {
344                LogError::Serde(format!("waitpoint wait encode error: {error}"))
345            })?,
346        )
347        .with_headers(wait_headers(&record.wait_id, &record.waitpoint_ids)),
348    )
349    .await
350    .map(|_| ())
351}
352
353pub async fn find_wait_terminal(
354    log: &Arc<AnyEventLog>,
355    wait_id: &str,
356) -> Result<Option<WaitpointWaitRecord>, LogError> {
357    let events = log.read_range(&waits_topic()?, None, usize::MAX).await?;
358    let mut latest = None;
359    for (_, event) in events {
360        if !matches!(
361            event.kind.as_str(),
362            "waitpoint_wait_completed"
363                | "waitpoint_wait_cancelled"
364                | "waitpoint_wait_timed_out"
365                | "waitpoint_wait_interrupted"
366        ) {
367            continue;
368        }
369        if event.headers.get("wait_id").map(String::as_str) != Some(wait_id) {
370            continue;
371        }
372        let Ok(record) = serde_json::from_value::<WaitpointWaitRecord>(event.payload) else {
373            continue;
374        };
375        latest = Some(record);
376    }
377    Ok(latest)
378}
379
380async fn append_waitpoint_state(
381    log: &Arc<AnyEventLog>,
382    kind: &str,
383    record: &WaitpointRecord,
384) -> Result<(), LogError> {
385    log.append(
386        &waitpoint_topic(&record.id)?,
387        LogEvent::new(
388            kind,
389            serde_json::to_value(record)
390                .map_err(|error| LogError::Serde(format!("waitpoint encode error: {error}")))?,
391        )
392        .with_headers(waitpoint_headers(record)),
393    )
394    .await
395    .map(|_| ())
396}
397
398fn wait_headers(wait_id: &str, waitpoint_ids: &[String]) -> BTreeMap<String, String> {
399    let mut headers = BTreeMap::new();
400    headers.insert("wait_id".to_string(), wait_id.to_string());
401    headers.insert("waitpoints".to_string(), waitpoint_ids.join(","));
402    headers
403}
404
405fn waitpoint_headers(record: &WaitpointRecord) -> BTreeMap<String, String> {
406    let mut headers = BTreeMap::new();
407    headers.insert("waitpoint_id".to_string(), record.id.clone());
408    headers.insert("status".to_string(), record.status.as_str().to_string());
409    if let Some(created_by) = record.created_by.as_ref() {
410        headers.insert("created_by".to_string(), created_by.clone());
411    }
412    if let Some(completed_by) = record.completed_by.as_ref() {
413        headers.insert("completed_by".to_string(), completed_by.clone());
414    }
415    if let Some(cancelled_by) = record.cancelled_by.as_ref() {
416        headers.insert("cancelled_by".to_string(), cancelled_by.clone());
417    }
418    headers
419}
420
421fn now_rfc3339() -> String {
422    // Was a second `now_utc().to_string()` on the unreachable format error,
423    // which is not RFC3339 at all; the shared helper emits the epoch instead.
424    harn_clock::system_now_rfc3339()
425}
426
427#[cfg(test)]
428mod tests {
429    use super::*;
430    use crate::event_log::{FileEventLog, MemoryEventLog};
431
432    #[tokio::test]
433    async fn waitpoint_state_persists_across_file_reopen() {
434        let dir = tempfile::tempdir().expect("tempdir");
435        let first = Arc::new(AnyEventLog::File(
436            FileEventLog::open(dir.path().to_path_buf(), 32).expect("open file log"),
437        ));
438        create_waitpoint(&first, "demo", Some("creator".to_string()), BTreeMap::new())
439            .await
440            .expect("create waitpoint");
441        complete_waitpoint(&first, "demo", Some("completer".to_string()))
442            .await
443            .expect("complete waitpoint");
444
445        let reopened = Arc::new(AnyEventLog::File(
446            FileEventLog::open(dir.path().to_path_buf(), 32).expect("reopen file log"),
447        ));
448        let state = load_waitpoint(&reopened, "demo")
449            .await
450            .expect("load state")
451            .expect("waitpoint exists");
452        assert_eq!(state.status, WaitpointStatus::Completed);
453        assert_eq!(state.completed_by.as_deref(), Some("completer"));
454    }
455
456    #[tokio::test]
457    async fn wait_terminal_lookup_returns_latest_terminal_record() {
458        let log = Arc::new(AnyEventLog::Memory(MemoryEventLog::new(32)));
459        append_wait_started(
460            &log,
461            &WaitpointWaitStartRecord {
462                wait_id: "wait-demo".to_string(),
463                waitpoint_ids: vec!["a".to_string(), "b".to_string()],
464                started_at: "2026-01-01T00:00:00Z".to_string(),
465                trace_id: Some("trace-demo".to_string()),
466                replay_of_event_id: None,
467            },
468        )
469        .await
470        .expect("append wait start");
471        append_wait_terminal(
472            &log,
473            &WaitpointWaitRecord {
474                wait_id: "wait-demo".to_string(),
475                waitpoint_ids: vec!["a".to_string(), "b".to_string()],
476                status: WaitpointWaitStatus::TimedOut,
477                started_at: "2026-01-01T00:00:00Z".to_string(),
478                resolved_at: "2026-01-01T00:01:00Z".to_string(),
479                waitpoints: Vec::new(),
480                cancelled_waitpoint_id: None,
481                trace_id: Some("trace-demo".to_string()),
482                replay_of_event_id: None,
483                reason: Some("deadline elapsed".to_string()),
484            },
485        )
486        .await
487        .expect("append wait result");
488
489        let record = find_wait_terminal(&log, "wait-demo")
490            .await
491            .expect("lookup wait result")
492            .expect("wait result exists");
493        assert_eq!(record.status, WaitpointWaitStatus::TimedOut);
494        assert_eq!(record.reason.as_deref(), Some("deadline elapsed"));
495    }
496}
harn_vm/waitpoints.rs

harn_vm/
waitpoints.rs
