lodestone_client/
bulk.rs

1use std::error::Error;
2use std::fmt::{self, Display, Formatter};
3
4use lodestone_wire::{
5    BulkDocumentItem, BulkDocumentItemOutcome, BulkDocumentOutcome, Digest, PrepareDocumentBulkRequest,
6    ResolveDocumentBulkRequest, Resource, SubmissionState, SubmitDocumentBulkRequest,
7};
8
9use crate::{CallOptions, Client, ClientError, SubmissionDisposition};
10
11#[derive(Clone, Debug, PartialEq)]
12pub struct BulkExecution {
13    pub outcome: BulkDocumentOutcome,
14    pub recovered: bool,
15}
16
17#[derive(Debug)]
18pub enum BulkRecoveryError {
19    InvalidInput,
20    InvalidCoreOutcome,
21    Client(ClientError),
22    Indeterminate(Box<BulkDocumentOutcome>),
23}
24
25impl Client {
26    /// Prepares every item, submits the batch at most once, and resolves any
27    /// ambiguous or partial outcome by the stable per-item locators.
28    pub async fn execute_document_bulk(
29        &self,
30        batch_id: Digest,
31        index: Resource,
32        items: Vec<BulkDocumentItem>,
33        options: CallOptions,
34    ) -> Result<BulkExecution, BulkRecoveryError> {
35        validate_bulk_input(&batch_id, &index, &items)?;
36        let prepared = self
37            .prepare_document_bulk(
38                PrepareDocumentBulkRequest {
39                    batch_id: batch_id.clone(),
40                    index: index.clone(),
41                    items: items.clone(),
42                },
43                options.clone(),
44            )
45            .await
46            .map_err(BulkRecoveryError::Client)?
47            .outcome;
48        validate_bulk_identity(&prepared, &batch_id, &index, &items)?;
49        if !prepared.complete
50            || prepared.items.iter().any(|item| {
51                item.state != SubmissionState::NotSubmitted
52                    || item.locator.is_none()
53                    || item.sequence.is_some()
54                    || item.error.is_some()
55            })
56        {
57            return Err(BulkRecoveryError::InvalidCoreOutcome);
58        }
59
60        let submitted = self
61            .submit_document_bulk(
62                SubmitDocumentBulkRequest {
63                    batch_id: batch_id.clone(),
64                    index: index.clone(),
65                    items: items.clone(),
66                    preparations: prepared.items.clone(),
67                },
68                options.clone(),
69            )
70            .await;
71        let (mut outcome, recovered) = match submitted {
72            Ok(result) => {
73                validate_bulk_identity(&result.outcome, &batch_id, &index, &items)?;
74                validate_submit_completion(&result.outcome)?;
75                if result.outcome.complete {
76                    return Ok(BulkExecution {
77                        outcome: result.outcome,
78                        recovered: false,
79                    });
80                }
81                (result.outcome, true)
82            }
83            Err(error) if error.submission_disposition() != SubmissionDisposition::NotSubmitted => (prepared, true),
84            Err(error) => return Err(BulkRecoveryError::Client(error)),
85        };
86
87        let maximum_resolutions = usize::from(self.pool_config().maximum_read_retries).saturating_add(1);
88        for _ in 0..maximum_resolutions {
89            let previous = outcome.clone();
90            let resolution = self
91                .resolve_document_bulk(
92                    ResolveDocumentBulkRequest {
93                        batch_id: batch_id.clone(),
94                        index: index.clone(),
95                        // Keep the last authoritative observation intact so a
96                        // deadline can return its full indeterminate outcome.
97                        items: outcome.items.clone(),
98                    },
99                    options.for_resolution(),
100                )
101                .await;
102            outcome = match resolution {
103                Ok(result) => result.outcome,
104                Err(error) if error.is_deadline_exceeded() => {
105                    return Err(BulkRecoveryError::Indeterminate(Box::new(outcome)));
106                }
107                Err(error) => return Err(BulkRecoveryError::Client(error)),
108            };
109            validate_bulk_identity(&outcome, &batch_id, &index, &items)?;
110            validate_resolution_completion(&outcome)?;
111            if outcome.complete {
112                return Ok(BulkExecution { outcome, recovered });
113            }
114            if outcome == previous {
115                break;
116            }
117        }
118        Err(BulkRecoveryError::Indeterminate(Box::new(outcome)))
119    }
120}
121
122fn validate_bulk_input(
123    batch_id: &Digest,
124    index: &Resource,
125    items: &[BulkDocumentItem],
126) -> Result<(), BulkRecoveryError> {
127    if batch_id.validate().is_err()
128        || index.kind != "index"
129        || index.name.is_empty()
130        || index.incarnation.validate().is_err()
131        || items.is_empty()
132        || items.len() > 1_000
133        || items.iter().any(|item| item.item_action_id.validate().is_err())
134    {
135        return Err(BulkRecoveryError::InvalidInput);
136    }
137    let mut action_ids = std::collections::BTreeSet::new();
138    if items
139        .iter()
140        .any(|item| !action_ids.insert(item.item_action_id.value.as_str()))
141    {
142        return Err(BulkRecoveryError::InvalidInput);
143    }
144    Ok(())
145}
146
147fn validate_bulk_identity(
148    outcome: &BulkDocumentOutcome,
149    batch_id: &Digest,
150    index: &Resource,
151    items: &[BulkDocumentItem],
152) -> Result<(), BulkRecoveryError> {
153    if &outcome.batch_id != batch_id || &outcome.index != index || outcome.items.len() != items.len() {
154        return Err(BulkRecoveryError::InvalidCoreOutcome);
155    }
156    for (expected, observed) in items.iter().zip(&outcome.items) {
157        if observed.item_action_id != expected.item_action_id || !valid_item_outcome(observed) {
158            return Err(BulkRecoveryError::InvalidCoreOutcome);
159        }
160    }
161    Ok(())
162}
163
164fn valid_item_outcome(item: &BulkDocumentItemOutcome) -> bool {
165    item.item_action_id.validate().is_ok()
166        && match item.state {
167            SubmissionState::NotSubmitted => item.sequence.is_none(),
168            SubmissionState::Submitted | SubmissionState::Indeterminate => item.locator.is_some(),
169            SubmissionState::Resolved => item.locator.is_some(),
170            SubmissionState::KnownRetired => item.locator.is_some() && item.error.is_some(),
171        }
172}
173
174fn validate_submit_completion(outcome: &BulkDocumentOutcome) -> Result<(), BulkRecoveryError> {
175    let fully_resolved = outcome.items.iter().all(|item| item.state == SubmissionState::Resolved);
176    if outcome.complete != fully_resolved {
177        return Err(BulkRecoveryError::InvalidCoreOutcome);
178    }
179    Ok(())
180}
181
182fn validate_resolution_completion(outcome: &BulkDocumentOutcome) -> Result<(), BulkRecoveryError> {
183    let terminal = outcome
184        .items
185        .iter()
186        .all(|item| matches!(item.state, SubmissionState::NotSubmitted | SubmissionState::Resolved));
187    if outcome.complete != terminal {
188        return Err(BulkRecoveryError::InvalidCoreOutcome);
189    }
190    Ok(())
191}
192
193impl Display for BulkRecoveryError {
194    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
195        formatter.write_str(match self {
196            Self::InvalidInput => "bulk request is not a valid current Lodestone API batch",
197            Self::InvalidCoreOutcome => "Core changed the bulk identity, order, or submission shape",
198            Self::Client(_) => "Core bulk execution failed",
199            Self::Indeterminate(_) => "Core bulk outcome remains indeterminate after bounded resolution",
200        })
201    }
202}
203
204impl Error for BulkRecoveryError {}
205
206#[cfg(test)]
207mod tests {
208    use lodestone_wire::{Digest, Identity, Locator, Resource};
209
210    use super::*;
211
212    fn outcome(states: &[SubmissionState], complete: bool) -> BulkDocumentOutcome {
213        BulkDocumentOutcome {
214            batch_id: Digest { value: "a".repeat(64) },
215            index: Resource {
216                kind: "index".to_owned(),
217                name: "events".to_owned(),
218                incarnation: Identity { value: "b".repeat(32) },
219            },
220            items: states
221                .iter()
222                .enumerate()
223                .map(|(ordinal, state)| BulkDocumentItemOutcome {
224                    item_action_id: Digest {
225                        value: format!("{ordinal:064x}"),
226                    },
227                    locator: Some(Locator {
228                        authority: "document".to_owned(),
229                        value: format!("locator-{ordinal}"),
230                    }),
231                    state: *state,
232                    sequence: None,
233                    mapping_generation: None,
234                    mapping_adopted: false,
235                    no_document_effect: false,
236                    error: None,
237                })
238                .collect(),
239            complete,
240        }
241    }
242
243    #[test]
244    fn partial_submit_is_recovered_instead_of_rejected_as_a_false_completion() {
245        let partial = outcome(&[SubmissionState::Resolved, SubmissionState::NotSubmitted], false);
246        assert!(validate_submit_completion(&partial).is_ok());
247        assert!(validate_resolution_completion(&partial).is_err());
248
249        let false_complete = outcome(&[SubmissionState::NotSubmitted], true);
250        assert!(validate_submit_completion(&false_complete).is_err());
251    }
252}
lodestone_client/bulk.rs

lodestone_client/
bulk.rs
