lodestone_client/
provisioning.rs

1//! Closed, secret-free recovery model for retained append provisioning.
2
3use aws_lc_rs::digest;
4use lodestone_wire::{
5    Digest, Identity, IndexResourceSpec, LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION, Locator, MappingMode,
6    PrepareCreateIndexRequest, Resource,
7};
8use serde::{Deserialize, Serialize};
9
10use crate::{
11    LocalOperatorWorkflowKind, ValidatedConnectionProfile, WorkflowJournalError, WorkflowJournalIntent,
12    WorkflowJournalTransition,
13};
14
15pub const PROVISION_RETAINED_APPEND_FORMAT_VERSION: u16 = 1;
16const MAX_NAME_BYTES: usize = 255;
17const MAX_ROLE_NAME_BYTES: usize = 128;
18
19/// Immutable, secret-free inputs identifying one provisioning workflow.
20#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
21#[serde(deny_unknown_fields)]
22pub struct ProvisionRetainedAppendIntent {
23    pub format_version: u16,
24    pub cluster_id: Identity,
25    pub collection_id: Identity,
26    pub admin_profile_sha256: [u8; 32],
27    pub consumer_profile_sha256: [u8; 32],
28    pub trust_sha256: [u8; 32],
29    pub definition_sha256: [u8; 32],
30    pub index_name: String,
31    pub role_name: String,
32}
33
34impl ProvisionRetainedAppendIntent {
35    pub fn new(
36        cluster_id: Identity,
37        collection_id: Identity,
38        admin_profile: &ValidatedConnectionProfile,
39        consumer_profile: &ValidatedConnectionProfile,
40        trust_sha256: [u8; 32],
41        definition: &PrepareCreateIndexRequest,
42        role_name: String,
43    ) -> Result<Self, WorkflowJournalError> {
44        let intent = Self {
45            format_version: PROVISION_RETAINED_APPEND_FORMAT_VERSION,
46            cluster_id,
47            collection_id,
48            admin_profile_sha256: profile_sha256(admin_profile)?,
49            consumer_profile_sha256: profile_sha256(consumer_profile)?,
50            trust_sha256,
51            definition_sha256: provision_definition_sha256(definition)?,
52            index_name: definition.name.clone(),
53            role_name,
54        };
55        intent.validate()?;
56        Ok(intent)
57    }
58}
59
60impl crate::workflow_journal::sealed::Sealed for ProvisionRetainedAppendIntent {}
61
62impl WorkflowJournalIntent for ProvisionRetainedAppendIntent {
63    type Transition = ProvisionRetainedAppendTransition;
64
65    const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::ProvisionRetainedAppend;
66
67    fn validate(&self) -> Result<(), WorkflowJournalError> {
68        if self.format_version != PROVISION_RETAINED_APPEND_FORMAT_VERSION
69            || !valid_identity(&self.cluster_id)
70            || !valid_identity(&self.collection_id)
71            || !valid_sha256(&self.admin_profile_sha256)
72            || !valid_sha256(&self.consumer_profile_sha256)
73            || !valid_sha256(&self.trust_sha256)
74            || !valid_sha256(&self.definition_sha256)
75            || !valid_index_name(&self.index_name)
76            || !valid_name(&self.role_name, MAX_ROLE_NAME_BYTES)
77        {
78            return Err(WorkflowJournalError::InvalidIntent);
79        }
80        Ok(())
81    }
82}
83
84/// How a prepared role action became known to have applied.
85#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
86#[serde(rename_all = "snake_case")]
87pub enum RoleAppendApplyResolution {
88    SubmitResponse,
89    Resolver,
90}
91
92/// Why one prepared locator is safe to stop submitting before a new prepare.
93#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
94#[serde(rename_all = "snake_case")]
95pub enum RoleAppendAttemptClosure {
96    ExpiredUnknown,
97    SupersededNoEffect,
98}
99
100/// Monotonic, secret-free observations committed by the workflow.
101#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
102#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
103pub enum ProvisionRetainedAppendTransition {
104    PreflightVerified,
105    CatalogPrepared {
106        locator: Locator,
107        semantic_digest: Digest,
108        expires_at: lodestone_wire::U64,
109        resource_spec: IndexResourceSpec,
110    },
111    CatalogApplied {
112        locator: Locator,
113        resource: Resource,
114    },
115    IndexActive {
116        resource: Resource,
117        index_revision: lodestone_wire::U64,
118    },
119    RolePrepared {
120        locator: Locator,
121        semantic_digest: Digest,
122        expires_at: lodestone_wire::U64,
123        predecessor_role_version: lodestone_wire::U64,
124        successor_role_version: lodestone_wire::U64,
125    },
126    RoleAttemptClosed {
127        locator: Locator,
128        closure: RoleAppendAttemptClosure,
129    },
130    RoleApplied {
131        locator: Locator,
132        resolution: RoleAppendApplyResolution,
133        target_policy_revision: lodestone_wire::U64,
134    },
135    RoleVerified {
136        resource: Resource,
137        role_version: lodestone_wire::U64,
138        #[serde(default, skip_serializing_if = "Option::is_none")]
139        target_policy_revision: Option<lodestone_wire::U64>,
140    },
141    DiscoveryVerified {
142        resource: Resource,
143        observed_policy_revision: lodestone_wire::U64,
144    },
145    Completed {
146        resource: Resource,
147        role_version: lodestone_wire::U64,
148        observed_policy_revision: lodestone_wire::U64,
149    },
150}
151
152#[derive(Clone, Debug, PartialEq)]
153pub struct ProvisionCatalogPreparation {
154    pub locator: Locator,
155    pub semantic_digest: Digest,
156    pub expires_at: lodestone_wire::U64,
157    pub resource_spec: IndexResourceSpec,
158}
159
160#[derive(Clone, Debug, PartialEq)]
161pub struct ProvisionRolePreparation {
162    pub locator: Locator,
163    pub semantic_digest: Digest,
164    pub expires_at: lodestone_wire::U64,
165    pub predecessor_role_version: lodestone_wire::U64,
166    pub successor_role_version: lodestone_wire::U64,
167}
168
169impl crate::workflow_journal::sealed::Sealed for ProvisionRetainedAppendTransition {}
170
171impl WorkflowJournalTransition for ProvisionRetainedAppendTransition {
172    fn validate(&self) -> Result<(), WorkflowJournalError> {
173        let valid = match self {
174            Self::PreflightVerified => true,
175            Self::CatalogPrepared {
176                locator,
177                semantic_digest,
178                expires_at,
179                resource_spec,
180            } => {
181                valid_locator(locator)
182                    && valid_digest(semantic_digest)
183                    && valid_u64(expires_at)
184                    && resource_spec.format_version == u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION)
185            }
186            Self::CatalogApplied { locator, resource } => valid_locator(locator) && valid_resource(resource),
187            Self::IndexActive {
188                resource,
189                index_revision,
190            } => valid_resource(resource) && valid_u64(index_revision),
191            Self::RolePrepared {
192                locator,
193                semantic_digest,
194                expires_at,
195                predecessor_role_version,
196                successor_role_version,
197            } => {
198                valid_locator(locator)
199                    && valid_digest(semantic_digest)
200                    && valid_u64(expires_at)
201                    && ordered_successor(predecessor_role_version, successor_role_version)
202            }
203            Self::RoleAttemptClosed { locator, .. } => valid_locator(locator),
204            Self::RoleApplied {
205                locator,
206                target_policy_revision,
207                ..
208            } => valid_locator(locator) && valid_u64(target_policy_revision),
209            Self::RoleVerified {
210                resource,
211                role_version,
212                target_policy_revision,
213            } => {
214                valid_resource(resource)
215                    && valid_u64(role_version)
216                    && target_policy_revision.as_ref().is_none_or(valid_u64)
217            }
218            Self::DiscoveryVerified {
219                resource,
220                observed_policy_revision,
221            } => valid_resource(resource) && valid_u64(observed_policy_revision),
222            Self::Completed {
223                resource,
224                role_version,
225                observed_policy_revision,
226            } => valid_resource(resource) && valid_u64(role_version) && valid_u64(observed_policy_revision),
227        };
228        if valid {
229            Ok(())
230        } else {
231            Err(WorkflowJournalError::InvalidTransition)
232        }
233    }
234}
235
236/// Replayed state after validating the complete transition ordering.
237#[derive(Clone, Debug, Default, PartialEq)]
238pub struct ProvisionRetainedAppendState {
239    preflight_verified: bool,
240    catalog_preparation: Option<ProvisionCatalogPreparation>,
241    catalog_resource: Option<Resource>,
242    index_active: bool,
243    role_preparation: Option<ProvisionRolePreparation>,
244    target_policy_revision: Option<lodestone_wire::U64>,
245    role_version: Option<lodestone_wire::U64>,
246    discovery_policy_revision: Option<lodestone_wire::U64>,
247    completed: bool,
248}
249
250impl ProvisionRetainedAppendState {
251    pub fn replay(transitions: &[ProvisionRetainedAppendTransition]) -> Result<Self, WorkflowJournalError> {
252        let mut state = Self::default();
253        for transition in transitions {
254            transition.validate()?;
255            state.apply(transition)?;
256        }
257        Ok(state)
258    }
259
260    fn apply(&mut self, transition: &ProvisionRetainedAppendTransition) -> Result<(), WorkflowJournalError> {
261        if self.completed {
262            return Err(WorkflowJournalError::Corrupt);
263        }
264        match transition {
265            ProvisionRetainedAppendTransition::PreflightVerified
266                if !self.preflight_verified && self.catalog_preparation.is_none() =>
267            {
268                self.preflight_verified = true;
269            }
270            ProvisionRetainedAppendTransition::CatalogPrepared {
271                locator,
272                semantic_digest,
273                expires_at,
274                resource_spec,
275            } if self.preflight_verified && self.catalog_preparation.is_none() && self.catalog_resource.is_none() => {
276                self.catalog_preparation = Some(ProvisionCatalogPreparation {
277                    locator: locator.clone(),
278                    semantic_digest: semantic_digest.clone(),
279                    expires_at: expires_at.clone(),
280                    resource_spec: resource_spec.clone(),
281                });
282            }
283            ProvisionRetainedAppendTransition::CatalogApplied { locator, resource }
284                if self.catalog_resource.is_none()
285                    && self
286                        .catalog_preparation
287                        .as_ref()
288                        .is_some_and(|prepared| &prepared.locator == locator) =>
289            {
290                self.catalog_resource = Some(resource.clone());
291            }
292            ProvisionRetainedAppendTransition::IndexActive { resource, .. }
293                if !self.index_active && self.catalog_resource.as_ref() == Some(resource) =>
294            {
295                self.index_active = true;
296            }
297            ProvisionRetainedAppendTransition::RolePrepared {
298                locator,
299                semantic_digest,
300                expires_at,
301                predecessor_role_version,
302                successor_role_version,
303            } if self.index_active && self.role_preparation.is_none() && self.role_version.is_none() => {
304                self.role_preparation = Some(ProvisionRolePreparation {
305                    locator: locator.clone(),
306                    semantic_digest: semantic_digest.clone(),
307                    expires_at: expires_at.clone(),
308                    predecessor_role_version: predecessor_role_version.clone(),
309                    successor_role_version: successor_role_version.clone(),
310                });
311            }
312            ProvisionRetainedAppendTransition::RoleAttemptClosed { locator, .. }
313                if self
314                    .role_preparation
315                    .as_ref()
316                    .is_some_and(|prepared| &prepared.locator == locator)
317                    && self.target_policy_revision.is_none() =>
318            {
319                self.role_preparation = None;
320            }
321            ProvisionRetainedAppendTransition::RoleApplied {
322                locator,
323                target_policy_revision,
324                ..
325            } if self
326                .role_preparation
327                .as_ref()
328                .is_some_and(|prepared| &prepared.locator == locator)
329                && self.target_policy_revision.is_none() =>
330            {
331                self.target_policy_revision = Some(target_policy_revision.clone());
332            }
333            ProvisionRetainedAppendTransition::RoleVerified {
334                resource,
335                role_version,
336                target_policy_revision,
337            } if self.index_active
338                && self.catalog_resource.as_ref() == Some(resource)
339                && self.role_version.is_none()
340                && target_policy_revision == &self.target_policy_revision
341                && self.role_version_is_admissible(role_version) =>
342            {
343                self.role_version = Some(role_version.clone());
344            }
345            ProvisionRetainedAppendTransition::DiscoveryVerified {
346                resource,
347                observed_policy_revision,
348            } if self.catalog_resource.as_ref() == Some(resource)
349                && self.role_version.is_some()
350                && self.discovery_policy_revision.is_none()
351                && revision_at_least(observed_policy_revision, self.target_policy_revision.as_ref()) =>
352            {
353                self.discovery_policy_revision = Some(observed_policy_revision.clone());
354            }
355            ProvisionRetainedAppendTransition::Completed {
356                resource,
357                role_version,
358                observed_policy_revision,
359            } if self.catalog_resource.as_ref() == Some(resource)
360                && revision_at_least(role_version, self.role_version.as_ref())
361                && self.discovery_policy_revision.as_ref() == Some(observed_policy_revision) =>
362            {
363                self.role_version = Some(role_version.clone());
364                self.completed = true;
365            }
366            _ => return Err(WorkflowJournalError::Corrupt),
367        }
368        Ok(())
369    }
370
371    fn role_version_is_admissible(&self, observed: &lodestone_wire::U64) -> bool {
372        match &self.role_preparation {
373            Some(preparation) if self.target_policy_revision.is_some() => {
374                revision_at_least(observed, Some(&preparation.successor_role_version))
375            }
376            Some(_) => false,
377            None => self.target_policy_revision.is_none(),
378        }
379    }
380
381    #[must_use]
382    pub const fn preflight_verified(&self) -> bool {
383        self.preflight_verified
384    }
385
386    #[must_use]
387    pub fn catalog_preparation(&self) -> Option<&ProvisionCatalogPreparation> {
388        self.catalog_preparation.as_ref()
389    }
390
391    #[must_use]
392    pub fn resource(&self) -> Option<&Resource> {
393        self.catalog_resource.as_ref()
394    }
395
396    #[must_use]
397    pub const fn index_active(&self) -> bool {
398        self.index_active
399    }
400
401    #[must_use]
402    pub fn role_preparation(&self) -> Option<&ProvisionRolePreparation> {
403        self.role_preparation.as_ref()
404    }
405
406    #[must_use]
407    pub fn target_policy_revision(&self) -> Option<&lodestone_wire::U64> {
408        self.target_policy_revision.as_ref()
409    }
410
411    #[must_use]
412    pub fn role_version(&self) -> Option<&lodestone_wire::U64> {
413        self.role_version.as_ref()
414    }
415
416    #[must_use]
417    pub fn discovery_policy_revision(&self) -> Option<&lodestone_wire::U64> {
418        self.discovery_policy_revision.as_ref()
419    }
420
421    #[must_use]
422    pub const fn completed(&self) -> bool {
423        self.completed
424    }
425}
426
427pub fn profile_sha256(profile: &ValidatedConnectionProfile) -> Result<[u8; 32], WorkflowJournalError> {
428    let bytes = profile
429        .encode_current()
430        .map_err(|_| WorkflowJournalError::InvalidIntent)?;
431    Ok(sha256(&bytes))
432}
433
434pub fn provision_definition_sha256(definition: &PrepareCreateIndexRequest) -> Result<[u8; 32], WorkflowJournalError> {
435    if definition.mapping.mode != MappingMode::AppendLog || definition.retained_event_identity != Some(true) {
436        return Err(WorkflowJournalError::InvalidIntent);
437    }
438    let bytes = serde_json::to_vec(definition).map_err(|_| WorkflowJournalError::Codec)?;
439    Ok(sha256(&bytes))
440}
441
442fn sha256(bytes: &[u8]) -> [u8; 32] {
443    let observed = digest::digest(&digest::SHA256, bytes);
444    let mut output = [0; 32];
445    output.copy_from_slice(observed.as_ref());
446    output
447}
448
449fn valid_identity(value: &Identity) -> bool {
450    value.value.len() == 32
451        && value
452            .value
453            .bytes()
454            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
455        && u128::from_str_radix(&value.value, 16).is_ok_and(|value| value != 0)
456}
457
458fn valid_sha256(value: &[u8; 32]) -> bool {
459    value.iter().any(|byte| *byte != 0)
460}
461
462fn valid_name(value: &str, maximum: usize) -> bool {
463    !value.is_empty()
464        && value.len() <= maximum
465        && value
466            .bytes()
467            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
468}
469
470fn valid_index_name(value: &str) -> bool {
471    !value.is_empty()
472        && value.len() <= MAX_NAME_BYTES
473        && !value
474            .bytes()
475            .any(|byte| byte.is_ascii_control() || byte.is_ascii_whitespace())
476}
477
478fn valid_locator(value: &Locator) -> bool {
479    !value.authority.is_empty()
480        && value.authority.len() <= 128
481        && !value.value.is_empty()
482        && value.value.len() <= 512
483        && !value.authority.chars().any(char::is_control)
484        && !value.value.chars().any(char::is_control)
485}
486
487fn valid_digest(value: &Digest) -> bool {
488    value.value.len() == 64
489        && value
490            .value
491            .bytes()
492            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
493        && value.value.bytes().any(|byte| byte != b'0')
494}
495
496fn valid_resource(value: &Resource) -> bool {
497    value.kind == "index" && valid_index_name(&value.name) && valid_identity(&value.incarnation)
498}
499
500fn valid_u64(value: &lodestone_wire::U64) -> bool {
501    value
502        .value
503        .parse::<u64>()
504        .is_ok_and(|parsed| parsed.to_string() == value.value)
505}
506
507fn ordered_successor(predecessor: &lodestone_wire::U64, successor: &lodestone_wire::U64) -> bool {
508    match (predecessor.value.parse::<u64>(), successor.value.parse::<u64>()) {
509        (Ok(predecessor), Ok(successor)) => predecessor.checked_add(1) == Some(successor),
510        _ => false,
511    }
512}
513
514fn revision_at_least(observed: &lodestone_wire::U64, minimum: Option<&lodestone_wire::U64>) -> bool {
515    match (
516        observed.value.parse::<u64>(),
517        minimum.map(|value| value.value.parse::<u64>()),
518    ) {
519        (Ok(_), None) => valid_u64(observed),
520        (Ok(observed), Some(Ok(minimum))) => observed >= minimum,
521        _ => false,
522    }
523}
524
525#[cfg(test)]
526mod tests {
527    use super::*;
528    use lodestone_wire::{ResourceDurabilityClass, ResourceStorageClass, U64};
529
530    fn id(value: char) -> Identity {
531        Identity {
532            value: value.to_string().repeat(32),
533        }
534    }
535
536    fn digest(value: char) -> Digest {
537        Digest {
538            value: value.to_string().repeat(64),
539        }
540    }
541
542    fn locator(value: &str) -> Locator {
543        Locator {
544            authority: "catalog".to_owned(),
545            value: value.to_owned(),
546        }
547    }
548
549    fn resource() -> Resource {
550        Resource {
551            kind: "index".to_owned(),
552            name: "events".to_owned(),
553            incarnation: id('a'),
554        }
555    }
556
557    fn resource_spec() -> IndexResourceSpec {
558        IndexResourceSpec {
559            format_version: u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION),
560            revision: U64 { value: "1".to_owned() },
561            initial_shards: 1,
562            target_shard_bytes: U64 { value: "1".to_owned() },
563            durability: ResourceDurabilityClass::RegionalObject,
564            storage_class: ResourceStorageClass::Standard,
565            ram_admission_weight: 1,
566            ram_hard_max_bytes: U64 { value: "1".to_owned() },
567            nvme_admission_weight: 1,
568            nvme_hard_max_bytes: U64 { value: "1".to_owned() },
569            refresh_objective_millis: 1,
570            memtable_hard_max_bytes: U64 { value: "1".to_owned() },
571            seal_parallelism: 1,
572            merge_parallelism: 1,
573            vector: lodestone_wire::VectorResourcePolicy {
574                quality: lodestone_wire::VectorQualityObjective::Balanced,
575                indexing: lodestone_wire::VectorIndexingObjective::Balanced,
576                residency: None,
577                calibration: None,
578            },
579            lifecycle_binding: None,
580            encryption_required: true,
581            resolver_capability: None,
582        }
583    }
584
585    fn valid_chain() -> Vec<ProvisionRetainedAppendTransition> {
586        let resource = resource();
587        vec![
588            ProvisionRetainedAppendTransition::PreflightVerified,
589            ProvisionRetainedAppendTransition::CatalogPrepared {
590                locator: locator("catalog-1"),
591                semantic_digest: digest('b'),
592                expires_at: U64 { value: "10".to_owned() },
593                resource_spec: resource_spec(),
594            },
595            ProvisionRetainedAppendTransition::CatalogApplied {
596                locator: locator("catalog-1"),
597                resource: resource.clone(),
598            },
599            ProvisionRetainedAppendTransition::IndexActive {
600                resource: resource.clone(),
601                index_revision: U64 { value: "1".to_owned() },
602            },
603            ProvisionRetainedAppendTransition::RolePrepared {
604                locator: locator("role-1"),
605                semantic_digest: digest('c'),
606                expires_at: U64 { value: "20".to_owned() },
607                predecessor_role_version: U64 { value: "4".to_owned() },
608                successor_role_version: U64 { value: "5".to_owned() },
609            },
610            ProvisionRetainedAppendTransition::RoleApplied {
611                locator: locator("role-1"),
612                resolution: RoleAppendApplyResolution::Resolver,
613                target_policy_revision: U64 { value: "9".to_owned() },
614            },
615            ProvisionRetainedAppendTransition::RoleVerified {
616                resource: resource.clone(),
617                role_version: U64 { value: "6".to_owned() },
618                target_policy_revision: Some(U64 { value: "9".to_owned() }),
619            },
620            ProvisionRetainedAppendTransition::DiscoveryVerified {
621                resource: resource.clone(),
622                observed_policy_revision: U64 { value: "10".to_owned() },
623            },
624            ProvisionRetainedAppendTransition::Completed {
625                resource,
626                role_version: U64 { value: "6".to_owned() },
627                observed_policy_revision: U64 { value: "10".to_owned() },
628            },
629        ]
630    }
631
632    #[test]
633    fn reducer_accepts_monotonic_completed_chain() -> Result<(), WorkflowJournalError> {
634        let state = ProvisionRetainedAppendState::replay(&valid_chain())?;
635        assert!(state.completed());
636        assert_eq!(state.resource(), Some(&resource()));
637        Ok(())
638    }
639
640    #[test]
641    fn intent_and_transitions_reject_zero_authority_values() {
642        let intent = ProvisionRetainedAppendIntent {
643            format_version: PROVISION_RETAINED_APPEND_FORMAT_VERSION,
644            cluster_id: id('0'),
645            collection_id: id('a'),
646            admin_profile_sha256: [1; 32],
647            consumer_profile_sha256: [2; 32],
648            trust_sha256: [3; 32],
649            definition_sha256: [4; 32],
650            index_name: "events".to_owned(),
651            role_name: "consumer".to_owned(),
652        };
653        assert_eq!(intent.validate(), Err(WorkflowJournalError::InvalidIntent));
654
655        let mut intent = intent;
656        intent.cluster_id = id('a');
657        intent.definition_sha256 = [0; 32];
658        assert_eq!(intent.validate(), Err(WorkflowJournalError::InvalidIntent));
659
660        let transition = ProvisionRetainedAppendTransition::CatalogPrepared {
661            locator: locator("catalog-1"),
662            semantic_digest: digest('0'),
663            expires_at: U64 { value: "10".to_owned() },
664            resource_spec: resource_spec(),
665        };
666        assert_eq!(transition.validate(), Err(WorkflowJournalError::InvalidTransition));
667    }
668
669    #[test]
670    fn reducer_rejects_out_of_order_or_post_terminal_records() {
671        assert_eq!(
672            ProvisionRetainedAppendState::replay(&[
673                ProvisionRetainedAppendTransition::PreflightVerified,
674                ProvisionRetainedAppendTransition::IndexActive {
675                    resource: resource(),
676                    index_revision: U64 { value: "1".to_owned() },
677                }
678            ]),
679            Err(WorkflowJournalError::Corrupt)
680        );
681        let mut transitions = valid_chain();
682        transitions.push(ProvisionRetainedAppendTransition::PreflightVerified);
683        assert_eq!(
684            ProvisionRetainedAppendState::replay(&transitions),
685            Err(WorkflowJournalError::Corrupt)
686        );
687    }
688
689    #[test]
690    fn reducer_retires_only_the_exact_role_locator() {
691        let mut transitions = valid_chain();
692        transitions.truncate(5);
693        transitions.push(ProvisionRetainedAppendTransition::RoleAttemptClosed {
694            locator: locator("other"),
695            closure: RoleAppendAttemptClosure::ExpiredUnknown,
696        });
697        assert_eq!(
698            ProvisionRetainedAppendState::replay(&transitions),
699            Err(WorkflowJournalError::Corrupt)
700        );
701    }
702
703    #[test]
704    fn reducer_accepts_a_later_role_version_that_retains_the_scope() -> Result<(), WorkflowJournalError> {
705        let mut transitions = valid_chain();
706        let Some(ProvisionRetainedAppendTransition::Completed { role_version, .. }) = transitions.last_mut() else {
707            return Err(WorkflowJournalError::Corrupt);
708        };
709        *role_version = U64 { value: "7".to_owned() };
710        let state = ProvisionRetainedAppendState::replay(&transitions)?;
711        assert_eq!(state.role_version(), Some(&U64 { value: "7".to_owned() }));
712        Ok(())
713    }
714}
lodestone_client/provisioning.rs

lodestone_client/
provisioning.rs
