elasticctl_api/value_lists/
read_back.rs

1use super::plan::container_mismatches;
2use super::{
3    Action, Container, Evidence, ReadOf, StoredValues, ValueSet, decode_export, read_container,
4};
5use elasticctl_core::{Error, ErrorKind, Result, Transport, urlencode};
6use std::{
7    future::Future,
8    pin::Pin,
9    time::{Duration, Instant},
10};
11
12pub trait Clock {
13    fn now(&self) -> Duration;
14    fn pause(&self, duration: Duration) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>;
15}
16
17pub struct SystemClock {
18    start: Instant,
19}
20impl Default for SystemClock {
21    fn default() -> Self {
22        Self {
23            start: Instant::now(),
24        }
25    }
26}
27impl Clock for SystemClock {
28    fn now(&self) -> Duration {
29        self.start.elapsed()
30    }
31    fn pause(&self, duration: Duration) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
32        Box::pin(tokio::time::sleep(duration))
33    }
34}
35
36const BUDGET: Duration = Duration::from_secs(120);
37const INTERVAL: Duration = Duration::from_secs(1);
38
39pub(super) enum Expected<'a> {
40    Values(&'a ValueSet),
41    Container,
42}
43pub(super) struct ReadBack {
44    pub evidence: Evidence,
45    pub stored: Option<StoredValues>,
46    pub verified: bool,
47}
48
49pub(super) async fn settle<C: Clock>(
50    t: &Transport,
51    container: &Container,
52    expected: Expected<'_>,
53    clock: &C,
54    last_export_end: &mut Option<Duration>,
55) -> ReadBack {
56    let mut evidence = Evidence::new(Action::ReadBack, &container.id, &container.name);
57    evidence.of = Some(match expected {
58        Expected::Values(_) => ReadOf::Values,
59        Expected::Container => ReadOf::Container,
60    });
61    evidence.sent = None;
62    evidence.values = None;
63    evidence.observed = Some(false);
64    let deadline = clock.now() + BUDGET;
65    let mut last_end: Option<Duration> = if matches!(expected, Expected::Values(_)) {
66        *last_export_end
67    } else {
68        None
69    };
70    loop {
71        if let Some(end) = last_end {
72            let wait = (end + INTERVAL).saturating_sub(clock.now());
73            if clock.now() + wait >= deadline {
74                break;
75            }
76            if !wait.is_zero() {
77                clock.pause(wait).await;
78            }
79        }
80        let left = deadline.saturating_sub(clock.now());
81        if left.is_zero() {
82            break;
83        }
84        let read = tokio::time::timeout(left, observe(t, container, &expected)).await;
85        last_end = Some(clock.now());
86        if matches!(expected, Expected::Values(_)) {
87            *last_export_end = last_end;
88        }
89        match read {
90            Ok(Ok(observed)) => {
91                evidence.observed = Some(true);
92                evidence.read_error = None;
93                let (matches, stored) = match observed {
94                    Observation::Values(stored) => {
95                        let Expected::Values(expected) = &expected else {
96                            unreachable!()
97                        };
98                        let set = stored.set();
99                        evidence.missing = Some(expected.difference(&set).count());
100                        evidence.extra = Some(set.difference(expected).count());
101                        (set == **expected, Some(stored))
102                    }
103                    Observation::Container(fields) => {
104                        let matches = fields.is_empty();
105                        evidence.mismatched = Some(fields);
106                        (matches, None)
107                    }
108                };
109                if matches && clock.now() <= deadline {
110                    return ReadBack {
111                        evidence,
112                        stored,
113                        verified: true,
114                    };
115                }
116            }
117            Ok(Err(error)) => evidence.read_error = Some(error.into()),
118            Err(_) => {
119                evidence.read_error = Some(
120                    Error::new(ErrorKind::Timeout, "value-list read-back budget exhausted").into(),
121                );
122                break;
123            }
124        }
125        if clock.now() >= deadline {
126            break;
127        }
128    }
129    ReadBack {
130        evidence,
131        stored: None,
132        verified: false,
133    }
134}
135
136enum Observation {
137    Values(StoredValues),
138    Container(Vec<String>),
139}
140async fn observe(
141    t: &Transport,
142    container: &Container,
143    expected: &Expected<'_>,
144) -> Result<Observation> {
145    match expected {
146        Expected::Values(_) => {
147            let text = t
148                .post_text_once(
149                    &format!(
150                        "/api/lists/items/_export?list_id={}",
151                        urlencode(&container.id)
152                    ),
153                    None,
154                )
155                .await?;
156            Ok(Observation::Values(decode_export(
157                &container.list_type,
158                &text,
159            )?))
160        }
161        Expected::Container => Ok(Observation::Container(
162            match read_container(t, &container.id).await? {
163                Some(observed) => container_mismatches(container, &observed.container),
164                None => vec!["id".into()],
165            },
166        )),
167    }
168}
elasticctl_api/value_lists/read_back.rs

elasticctl_api/value_lists/
read_back.rs
