lodestone_client/
index_change.rs

1use std::error::Error;
2use std::fmt::{self, Display, Formatter};
3
4use lodestone_wire::{
5    ApplyIndexChangeRequest, IndexChangeApplyReceipt, IndexChangeResolutionState, ResolveIndexChangeRequest,
6    ResolveIndexChangeResult,
7};
8
9use crate::{CallOptions, Client, ClientError, SubmissionDisposition};
10
11#[derive(Clone, Debug, PartialEq)]
12pub struct IndexChangeExecution {
13    pub receipt: IndexChangeApplyReceipt,
14    pub recovered: bool,
15}
16
17#[derive(Clone, Debug, PartialEq)]
18pub struct IndexChangeRecovery {
19    request: ApplyIndexChangeRequest,
20    resolution: Option<ResolveIndexChangeResult>,
21}
22
23impl IndexChangeRecovery {
24    #[must_use]
25    pub const fn request(&self) -> &ApplyIndexChangeRequest {
26        &self.request
27    }
28
29    #[must_use]
30    pub const fn resolution(&self) -> Option<&ResolveIndexChangeResult> {
31        self.resolution.as_ref()
32    }
33}
34
35#[derive(Debug)]
36pub enum IndexChangeRecoveryError {
37    InvalidInput,
38    InvalidCoreOutcome,
39    /// The original apply call authoritatively did not submit. Retrying the
40    /// caller's exact request is admitted.
41    Client(ClientError),
42    /// The catalog authority requires the exact original action to be replayed.
43    ExactReplayRequired(Box<IndexChangeRecovery>),
44    /// A read-only resolver failed after submission may have occurred.
45    Indeterminate(Box<IndexChangeRecovery>),
46    Failed(Box<IndexChangeRecovery>),
47    KnownRetired(Box<IndexChangeRecovery>),
48    ExpiredUnknown(Box<IndexChangeRecovery>),
49}
50
51impl Client {
52    /// Applies one validated index change and resolves any response loss by the
53    /// original action identity. Success always carries the exact durable
54    /// receipt; a nonterminal result preserves the request needed for replay.
55    pub async fn execute_index_change(
56        &self,
57        request: ApplyIndexChangeRequest,
58        options: CallOptions,
59    ) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
60        validate_request(&request)?;
61        self.submit_exact_index_change(request, options, false).await
62    }
63
64    /// Replays only the exact request retained from a catalog-authoritative
65    /// `not_submitted`, `prepared`, or `pending` resolution.
66    pub async fn replay_index_change(
67        &self,
68        recovery: IndexChangeRecovery,
69        options: CallOptions,
70    ) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
71        validate_recovery(&recovery)?;
72        if !recovery.resolution.as_ref().is_some_and(|resolution| {
73            matches!(
74                resolution.state,
75                IndexChangeResolutionState::NotSubmitted
76                    | IndexChangeResolutionState::Prepared
77                    | IndexChangeResolutionState::Pending
78            )
79        }) {
80            return Err(IndexChangeRecoveryError::InvalidInput);
81        }
82        self.submit_exact_index_change(recovery.request, options, true).await
83    }
84
85    /// Continues read-only resolution after the prior resolver call failed.
86    pub async fn resume_index_change(
87        &self,
88        recovery: IndexChangeRecovery,
89        options: CallOptions,
90    ) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
91        validate_recovery(&recovery)?;
92        self.resolve_exact_index_change(recovery, options, true).await
93    }
94
95    async fn submit_exact_index_change(
96        &self,
97        request: ApplyIndexChangeRequest,
98        options: CallOptions,
99        recovered: bool,
100    ) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
101        match self.apply_index_change(request.clone(), options.clone()).await {
102            Ok(result) => {
103                validate_receipt(&request, &result.receipt)?;
104                Ok(IndexChangeExecution {
105                    receipt: result.receipt,
106                    recovered,
107                })
108            }
109            Err(error) if !recovered && error.submission_disposition() == SubmissionDisposition::NotSubmitted => {
110                Err(IndexChangeRecoveryError::Client(error))
111            }
112            Err(_) => {
113                self.resolve_exact_index_change(
114                    IndexChangeRecovery {
115                        request,
116                        resolution: None,
117                    },
118                    options,
119                    true,
120                )
121                .await
122            }
123        }
124    }
125
126    async fn resolve_exact_index_change(
127        &self,
128        mut recovery: IndexChangeRecovery,
129        options: CallOptions,
130        recovered: bool,
131    ) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
132        let resolution = self
133            .resolve_index_change(resolve_request(&recovery.request), options.for_resolution())
134            .await;
135        recovery.resolution = match resolution {
136            Ok(resolution) => Some(resolution),
137            Err(_) => return Err(IndexChangeRecoveryError::Indeterminate(Box::new(recovery))),
138        };
139        classify_resolution(recovery, recovered)
140    }
141}
142
143fn resolve_request(request: &ApplyIndexChangeRequest) -> ResolveIndexChangeRequest {
144    ResolveIndexChangeRequest {
145        action_id: request.action_id.clone(),
146        index: request.index.clone(),
147        expected_revision: request.expected_revision.clone(),
148        mapping: request.mapping.clone(),
149        policies: request.policies.clone(),
150    }
151}
152
153fn validate_request(request: &ApplyIndexChangeRequest) -> Result<(), IndexChangeRecoveryError> {
154    if request.action_id.validate().is_err()
155        || request.index.kind != "index"
156        || request.index.name.is_empty()
157        || request.index.incarnation.validate().is_err()
158        || request.expected_revision.validate().is_err()
159        || (request.mapping.is_some() == request.policies.is_some())
160    {
161        return Err(IndexChangeRecoveryError::InvalidInput);
162    }
163    if request
164        .mapping
165        .as_ref()
166        .is_some_and(|mapping| mapping.generation.validate().is_err())
167    {
168        return Err(IndexChangeRecoveryError::InvalidInput);
169    }
170    Ok(())
171}
172
173fn validate_recovery(recovery: &IndexChangeRecovery) -> Result<(), IndexChangeRecoveryError> {
174    validate_request(&recovery.request)?;
175    if let Some(resolution) = &recovery.resolution {
176        validate_resolution_shape(resolution)?;
177        if let Some(receipt) = &resolution.receipt {
178            validate_receipt(&recovery.request, receipt)?;
179        }
180    }
181    Ok(())
182}
183
184fn validate_receipt(
185    request: &ApplyIndexChangeRequest,
186    receipt: &IndexChangeApplyReceipt,
187) -> Result<(), IndexChangeRecoveryError> {
188    let expected_revision = request
189        .expected_revision
190        .value
191        .parse::<u64>()
192        .ok()
193        .and_then(|revision| revision.checked_add(1))
194        .map(|revision| revision.to_string())
195        .ok_or(IndexChangeRecoveryError::InvalidCoreOutcome)?;
196    let expected_state = if request.mapping.is_some() {
197        "mapping_applied"
198    } else {
199        "policy_applied"
200    };
201    if receipt.action_id != request.action_id
202        || receipt.semantic_digest != request.action_id
203        || receipt.index != request.index
204        || receipt.revision.value != expected_revision
205        || receipt.state != expected_state
206        || receipt.schema_generation.validate().is_err()
207        || request
208            .mapping
209            .as_ref()
210            .is_some_and(|mapping| mapping.generation != receipt.schema_generation)
211    {
212        return Err(IndexChangeRecoveryError::InvalidCoreOutcome);
213    }
214    Ok(())
215}
216
217fn validate_resolution_shape(resolution: &ResolveIndexChangeResult) -> Result<(), IndexChangeRecoveryError> {
218    let valid = match resolution.state {
219        IndexChangeResolutionState::Applied => resolution.receipt.is_some() && resolution.failure.is_none(),
220        IndexChangeResolutionState::Failed => resolution.receipt.is_none() && resolution.failure.is_some(),
221        IndexChangeResolutionState::NotSubmitted
222        | IndexChangeResolutionState::Prepared
223        | IndexChangeResolutionState::Pending
224        | IndexChangeResolutionState::KnownRetired
225        | IndexChangeResolutionState::ExpiredUnknown => resolution.receipt.is_none() && resolution.failure.is_none(),
226    };
227    valid.then_some(()).ok_or(IndexChangeRecoveryError::InvalidCoreOutcome)
228}
229
230fn classify_resolution(
231    recovery: IndexChangeRecovery,
232    recovered: bool,
233) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
234    validate_recovery(&recovery)?;
235    let resolution = recovery
236        .resolution
237        .as_ref()
238        .ok_or(IndexChangeRecoveryError::InvalidCoreOutcome)?;
239    match resolution.state {
240        IndexChangeResolutionState::Applied => Ok(IndexChangeExecution {
241            receipt: resolution
242                .receipt
243                .clone()
244                .ok_or(IndexChangeRecoveryError::InvalidCoreOutcome)?,
245            recovered,
246        }),
247        IndexChangeResolutionState::NotSubmitted
248        | IndexChangeResolutionState::Prepared
249        | IndexChangeResolutionState::Pending => Err(IndexChangeRecoveryError::ExactReplayRequired(Box::new(recovery))),
250        IndexChangeResolutionState::Failed => Err(IndexChangeRecoveryError::Failed(Box::new(recovery))),
251        IndexChangeResolutionState::KnownRetired => Err(IndexChangeRecoveryError::KnownRetired(Box::new(recovery))),
252        IndexChangeResolutionState::ExpiredUnknown => Err(IndexChangeRecoveryError::ExpiredUnknown(Box::new(recovery))),
253    }
254}
255
256impl Display for IndexChangeRecoveryError {
257    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
258        formatter.write_str(match self {
259            Self::InvalidInput => "index change request is not a valid current Lodestone API action",
260            Self::InvalidCoreOutcome => "Core changed the index action identity or recovery shape",
261            Self::Client(_) => "Core index change was not submitted",
262            Self::ExactReplayRequired(_) => "Core index change requires exact replay of the original action",
263            Self::Indeterminate(_) => "Core index change outcome remains indeterminate",
264            Self::Failed(_) => "Core index change reached a terminal failed receipt",
265            Self::KnownRetired(_) => "Core index incarnation is authoritatively retired",
266            Self::ExpiredUnknown(_) => "Core index change outcome evidence expired",
267        })
268    }
269}
270
271impl Error for IndexChangeRecoveryError {
272    fn source(&self) -> Option<&(dyn Error + 'static)> {
273        match self {
274            Self::Client(error) => Some(error),
275            Self::InvalidInput
276            | Self::InvalidCoreOutcome
277            | Self::ExactReplayRequired(_)
278            | Self::Indeterminate(_)
279            | Self::Failed(_)
280            | Self::KnownRetired(_)
281            | Self::ExpiredUnknown(_) => None,
282        }
283    }
284}
285
286#[cfg(test)]
287mod tests {
288    use lodestone_wire::{Digest, Identity, Resource, U64};
289
290    use super::*;
291
292    fn request() -> ApplyIndexChangeRequest {
293        ApplyIndexChangeRequest {
294            action_id: Digest { value: "a".repeat(64) },
295            index: Resource {
296                kind: "index".to_owned(),
297                name: "documents".to_owned(),
298                incarnation: Identity { value: "b".repeat(32) },
299            },
300            expected_revision: U64 { value: "7".to_owned() },
301            mapping: None,
302            policies: Some(lodestone_wire::IndexPolicies {
303                durability: "regional_object".to_owned(),
304                storage: "standard".to_owned(),
305                cache: "cold_tolerant".to_owned(),
306                default_visibility: Some("applied".to_owned()),
307                field_promotion: lodestone_wire::FieldPromotionConfiguration {
308                    format_version: 3,
309                    enabled: false,
310                    minimum_query_count: 1,
311                    maximum_promoted_fields: 1,
312                    maximum_cardinality: U64 { value: "1".to_owned() },
313                    maximum_extra_bytes: U64 { value: "1".to_owned() },
314                    allow_sensitive: false,
315                },
316                overload: lodestone_wire::OverloadConfiguration {
317                    format_version: 3,
318                    mode: "reject".to_owned(),
319                    maximum_backlog_bytes: U64 { value: "1".to_owned() },
320                    maximum_backlog_age_millis: U64 { value: "1".to_owned() },
321                    maximum_collection_share_bytes: U64 { value: "1".to_owned() },
322                    maximum_recovery_ops_per_pass: 1,
323                    maximum_pending_overlay_documents: U64 { value: "1".to_owned() },
324                },
325            }),
326        }
327    }
328
329    #[test]
330    fn pending_resolution_requires_exact_identity_replay() -> Result<(), Box<dyn Error>> {
331        let recovery = IndexChangeRecovery {
332            request: request(),
333            resolution: Some(ResolveIndexChangeResult {
334                state: IndexChangeResolutionState::Pending,
335                receipt: None,
336                failure: None,
337            }),
338        };
339        match classify_resolution(recovery, true) {
340            Err(IndexChangeRecoveryError::ExactReplayRequired(recovery)) => {
341                assert_eq!(recovery.request().action_id.value, "a".repeat(64));
342                Ok(())
343            }
344            _ => Err("pending resolution did not require exact replay".into()),
345        }
346    }
347
348    #[test]
349    fn applied_resolution_requires_exact_successor_receipt() -> Result<(), Box<dyn Error>> {
350        let request = request();
351        let receipt = IndexChangeApplyReceipt {
352            action_id: request.action_id.clone(),
353            semantic_digest: request.action_id.clone(),
354            index: request.index.clone(),
355            revision: U64 { value: "8".to_owned() },
356            schema_generation: U64 { value: "3".to_owned() },
357            state: "policy_applied".to_owned(),
358        };
359        let execution = classify_resolution(
360            IndexChangeRecovery {
361                request,
362                resolution: Some(ResolveIndexChangeResult {
363                    state: IndexChangeResolutionState::Applied,
364                    receipt: Some(receipt.clone()),
365                    failure: None,
366                }),
367            },
368            true,
369        )?;
370        assert_eq!(execution.receipt, receipt);
371        assert!(execution.recovered);
372        Ok(())
373    }
374}
lodestone_client/index_change.rs

lodestone_client/
index_change.rs
