lodestone_client/
provisioning.rs

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

lodestone_client/
provisioning.rs
