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 = 2;
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    /// Exact catalog state adopted by a new version-2 local workflow. The
116    /// catalog read remains authoritative; this observation is not an action
117    /// receipt and cannot satisfy a version-1 journal.
118    CatalogAdopted {
119        resource: Resource,
120        resource_spec: IndexResourceSpec,
121    },
122    IndexActive {
123        resource: Resource,
124        index_revision: lodestone_wire::U64,
125    },
126    RolePrepared {
127        locator: Locator,
128        semantic_digest: Digest,
129        expires_at: lodestone_wire::U64,
130        predecessor_role_version: lodestone_wire::U64,
131        successor_role_version: lodestone_wire::U64,
132    },
133    RoleAttemptClosed {
134        locator: Locator,
135        closure: RoleAppendAttemptClosure,
136    },
137    RoleApplied {
138        locator: Locator,
139        resolution: RoleAppendApplyResolution,
140        target_policy_revision: lodestone_wire::U64,
141    },
142    RoleVerified {
143        resource: Resource,
144        role_version: lodestone_wire::U64,
145        #[serde(default, skip_serializing_if = "Option::is_none")]
146        target_policy_revision: Option<lodestone_wire::U64>,
147    },
148    DiscoveryVerified {
149        resource: Resource,
150        observed_policy_revision: lodestone_wire::U64,
151    },
152    Completed {
153        resource: Resource,
154        role_version: lodestone_wire::U64,
155        observed_policy_revision: lodestone_wire::U64,
156    },
157}
158
159#[derive(Clone, Debug, PartialEq)]
160pub struct ProvisionCatalogPreparation {
161    pub locator: Locator,
162    pub semantic_digest: Digest,
163    pub expires_at: lodestone_wire::U64,
164    pub resource_spec: IndexResourceSpec,
165}
166
167#[derive(Clone, Debug, PartialEq)]
168pub struct ProvisionRolePreparation {
169    pub locator: Locator,
170    pub semantic_digest: Digest,
171    pub expires_at: lodestone_wire::U64,
172    pub predecessor_role_version: lodestone_wire::U64,
173    pub successor_role_version: lodestone_wire::U64,
174}
175
176impl crate::workflow_journal::sealed::Sealed for ProvisionRetainedAppendTransition {}
177
178impl WorkflowJournalTransition for ProvisionRetainedAppendTransition {
179    fn validate(&self) -> Result<(), WorkflowJournalError> {
180        let valid = match self {
181            Self::PreflightVerified => true,
182            Self::CatalogPrepared {
183                locator,
184                semantic_digest,
185                expires_at,
186                resource_spec,
187            } => {
188                valid_locator(locator)
189                    && valid_digest(semantic_digest)
190                    && valid_u64(expires_at)
191                    && resource_spec.format_version == u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION)
192            }
193            Self::CatalogApplied { locator, resource } => valid_locator(locator) && valid_resource(resource),
194            Self::CatalogAdopted {
195                resource,
196                resource_spec,
197            } => {
198                valid_resource(resource)
199                    && resource_spec.format_version == u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION)
200            }
201            Self::IndexActive {
202                resource,
203                index_revision,
204            } => valid_resource(resource) && valid_u64(index_revision),
205            Self::RolePrepared {
206                locator,
207                semantic_digest,
208                expires_at,
209                predecessor_role_version,
210                successor_role_version,
211            } => {
212                valid_locator(locator)
213                    && valid_digest(semantic_digest)
214                    && valid_u64(expires_at)
215                    && ordered_successor(predecessor_role_version, successor_role_version)
216            }
217            Self::RoleAttemptClosed { locator, .. } => valid_locator(locator),
218            Self::RoleApplied {
219                locator,
220                target_policy_revision,
221                ..
222            } => valid_locator(locator) && valid_u64(target_policy_revision),
223            Self::RoleVerified {
224                resource,
225                role_version,
226                target_policy_revision,
227            } => {
228                valid_resource(resource)
229                    && valid_u64(role_version)
230                    && target_policy_revision.as_ref().is_none_or(valid_u64)
231            }
232            Self::DiscoveryVerified {
233                resource,
234                observed_policy_revision,
235            } => valid_resource(resource) && valid_u64(observed_policy_revision),
236            Self::Completed {
237                resource,
238                role_version,
239                observed_policy_revision,
240            } => valid_resource(resource) && valid_u64(role_version) && valid_u64(observed_policy_revision),
241        };
242        if valid {
243            Ok(())
244        } else {
245            Err(WorkflowJournalError::InvalidTransition)
246        }
247    }
248}
249
250/// Replayed state after validating the complete transition ordering.
251#[derive(Clone, Debug, Default, PartialEq)]
252pub struct ProvisionRetainedAppendState {
253    preflight_verified: bool,
254    catalog_preparation: Option<ProvisionCatalogPreparation>,
255    catalog_resource: Option<Resource>,
256    adopted_resource_spec: Option<IndexResourceSpec>,
257    index_active: bool,
258    role_preparation: Option<ProvisionRolePreparation>,
259    target_policy_revision: Option<lodestone_wire::U64>,
260    role_version: Option<lodestone_wire::U64>,
261    discovery_policy_revision: Option<lodestone_wire::U64>,
262    completed: bool,
263}
264
265impl ProvisionRetainedAppendState {
266    pub fn replay(transitions: &[ProvisionRetainedAppendTransition]) -> Result<Self, WorkflowJournalError> {
267        let mut state = Self::default();
268        for transition in transitions {
269            transition.validate()?;
270            state.apply(transition)?;
271        }
272        Ok(state)
273    }
274
275    fn apply(&mut self, transition: &ProvisionRetainedAppendTransition) -> Result<(), WorkflowJournalError> {
276        if self.completed {
277            return Err(WorkflowJournalError::Corrupt);
278        }
279        match transition {
280            ProvisionRetainedAppendTransition::PreflightVerified
281                if !self.preflight_verified && self.catalog_preparation.is_none() =>
282            {
283                self.preflight_verified = true;
284            }
285            ProvisionRetainedAppendTransition::CatalogPrepared {
286                locator,
287                semantic_digest,
288                expires_at,
289                resource_spec,
290            } if self.preflight_verified && self.catalog_preparation.is_none() && self.catalog_resource.is_none() => {
291                self.catalog_preparation = Some(ProvisionCatalogPreparation {
292                    locator: locator.clone(),
293                    semantic_digest: semantic_digest.clone(),
294                    expires_at: expires_at.clone(),
295                    resource_spec: resource_spec.clone(),
296                });
297            }
298            ProvisionRetainedAppendTransition::CatalogApplied { locator, resource }
299                if self.catalog_resource.is_none()
300                    && self
301                        .catalog_preparation
302                        .as_ref()
303                        .is_some_and(|prepared| &prepared.locator == locator) =>
304            {
305                self.catalog_resource = Some(resource.clone());
306            }
307            ProvisionRetainedAppendTransition::CatalogAdopted {
308                resource,
309                resource_spec,
310            } if self.preflight_verified
311                && self.catalog_preparation.is_none()
312                && self.catalog_resource.is_none()
313                && self.adopted_resource_spec.is_none() =>
314            {
315                self.catalog_resource = Some(resource.clone());
316                self.adopted_resource_spec = Some(resource_spec.clone());
317            }
318            ProvisionRetainedAppendTransition::IndexActive { resource, .. }
319                if !self.index_active && self.catalog_resource.as_ref() == Some(resource) =>
320            {
321                self.index_active = true;
322            }
323            ProvisionRetainedAppendTransition::RolePrepared {
324                locator,
325                semantic_digest,
326                expires_at,
327                predecessor_role_version,
328                successor_role_version,
329            } if self.index_active && self.role_preparation.is_none() && self.role_version.is_none() => {
330                self.role_preparation = Some(ProvisionRolePreparation {
331                    locator: locator.clone(),
332                    semantic_digest: semantic_digest.clone(),
333                    expires_at: expires_at.clone(),
334                    predecessor_role_version: predecessor_role_version.clone(),
335                    successor_role_version: successor_role_version.clone(),
336                });
337            }
338            ProvisionRetainedAppendTransition::RoleAttemptClosed { locator, .. }
339                if self
340                    .role_preparation
341                    .as_ref()
342                    .is_some_and(|prepared| &prepared.locator == locator)
343                    && self.target_policy_revision.is_none() =>
344            {
345                self.role_preparation = None;
346            }
347            ProvisionRetainedAppendTransition::RoleApplied {
348                locator,
349                target_policy_revision,
350                ..
351            } if self
352                .role_preparation
353                .as_ref()
354                .is_some_and(|prepared| &prepared.locator == locator)
355                && self.target_policy_revision.is_none() =>
356            {
357                self.target_policy_revision = Some(target_policy_revision.clone());
358            }
359            ProvisionRetainedAppendTransition::RoleVerified {
360                resource,
361                role_version,
362                target_policy_revision,
363            } if self.index_active
364                && self.catalog_resource.as_ref() == Some(resource)
365                && self.role_version.is_none()
366                && target_policy_revision == &self.target_policy_revision
367                && self.role_version_is_admissible(role_version) =>
368            {
369                self.role_version = Some(role_version.clone());
370            }
371            ProvisionRetainedAppendTransition::DiscoveryVerified {
372                resource,
373                observed_policy_revision,
374            } if self.catalog_resource.as_ref() == Some(resource)
375                && self.role_version.is_some()
376                && self.discovery_policy_revision.is_none()
377                && revision_at_least(observed_policy_revision, self.target_policy_revision.as_ref()) =>
378            {
379                self.discovery_policy_revision = Some(observed_policy_revision.clone());
380            }
381            ProvisionRetainedAppendTransition::Completed {
382                resource,
383                role_version,
384                observed_policy_revision,
385            } if self.catalog_resource.as_ref() == Some(resource)
386                && revision_at_least(role_version, self.role_version.as_ref())
387                && self.discovery_policy_revision.as_ref() == Some(observed_policy_revision) =>
388            {
389                self.role_version = Some(role_version.clone());
390                self.completed = true;
391            }
392            _ => return Err(WorkflowJournalError::Corrupt),
393        }
394        Ok(())
395    }
396
397    fn role_version_is_admissible(&self, observed: &lodestone_wire::U64) -> bool {
398        match &self.role_preparation {
399            Some(preparation) if self.target_policy_revision.is_some() => {
400                revision_at_least(observed, Some(&preparation.successor_role_version))
401            }
402            Some(_) => false,
403            None => self.target_policy_revision.is_none(),
404        }
405    }
406
407    #[must_use]
408    pub const fn preflight_verified(&self) -> bool {
409        self.preflight_verified
410    }
411
412    #[must_use]
413    pub fn catalog_preparation(&self) -> Option<&ProvisionCatalogPreparation> {
414        self.catalog_preparation.as_ref()
415    }
416
417    #[must_use]
418    pub fn resource(&self) -> Option<&Resource> {
419        self.catalog_resource.as_ref()
420    }
421
422    #[must_use]
423    pub fn resource_spec(&self) -> Option<&IndexResourceSpec> {
424        self.catalog_preparation
425            .as_ref()
426            .map(|preparation| &preparation.resource_spec)
427            .or(self.adopted_resource_spec.as_ref())
428    }
429
430    #[must_use]
431    pub const fn index_active(&self) -> bool {
432        self.index_active
433    }
434
435    #[must_use]
436    pub fn role_preparation(&self) -> Option<&ProvisionRolePreparation> {
437        self.role_preparation.as_ref()
438    }
439
440    #[must_use]
441    pub fn target_policy_revision(&self) -> Option<&lodestone_wire::U64> {
442        self.target_policy_revision.as_ref()
443    }
444
445    #[must_use]
446    pub fn role_version(&self) -> Option<&lodestone_wire::U64> {
447        self.role_version.as_ref()
448    }
449
450    #[must_use]
451    pub fn discovery_policy_revision(&self) -> Option<&lodestone_wire::U64> {
452        self.discovery_policy_revision.as_ref()
453    }
454
455    #[must_use]
456    pub const fn completed(&self) -> bool {
457        self.completed
458    }
459}
460
461pub fn profile_sha256(profile: &ValidatedConnectionProfile) -> Result<[u8; 32], WorkflowJournalError> {
462    let bytes = profile
463        .encode_current()
464        .map_err(|_| WorkflowJournalError::InvalidIntent)?;
465    Ok(sha256(&bytes))
466}
467
468pub fn provision_definition_sha256(definition: &PrepareCreateIndexRequest) -> Result<[u8; 32], WorkflowJournalError> {
469    if definition.mapping.mode != MappingMode::AppendLog || definition.retained_event_identity != Some(true) {
470        return Err(WorkflowJournalError::InvalidIntent);
471    }
472    let bytes = serde_json::to_vec(definition).map_err(|_| WorkflowJournalError::Codec)?;
473    Ok(sha256(&bytes))
474}
475
476fn sha256(bytes: &[u8]) -> [u8; 32] {
477    let observed = digest::digest(&digest::SHA256, bytes);
478    let mut output = [0; 32];
479    output.copy_from_slice(observed.as_ref());
480    output
481}
482
483fn valid_identity(value: &Identity) -> bool {
484    value.value.len() == 32
485        && value
486            .value
487            .bytes()
488            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
489        && u128::from_str_radix(&value.value, 16).is_ok_and(|value| value != 0)
490}
491
492fn valid_sha256(value: &[u8; 32]) -> bool {
493    value.iter().any(|byte| *byte != 0)
494}
495
496fn valid_name(value: &str, maximum: usize) -> bool {
497    !value.is_empty()
498        && value.len() <= maximum
499        && value
500            .bytes()
501            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
502}
503
504fn valid_index_name(value: &str) -> bool {
505    !value.is_empty()
506        && value.len() <= MAX_NAME_BYTES
507        && !value
508            .bytes()
509            .any(|byte| byte.is_ascii_control() || byte.is_ascii_whitespace())
510}
511
512fn valid_locator(value: &Locator) -> bool {
513    !value.authority.is_empty()
514        && value.authority.len() <= 128
515        && !value.value.is_empty()
516        && value.value.len() <= 512
517        && !value.authority.chars().any(char::is_control)
518        && !value.value.chars().any(char::is_control)
519}
520
521fn valid_digest(value: &Digest) -> bool {
522    value.value.len() == 64
523        && value
524            .value
525            .bytes()
526            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
527        && value.value.bytes().any(|byte| byte != b'0')
528}
529
530fn valid_resource(value: &Resource) -> bool {
531    value.kind == "index" && valid_index_name(&value.name) && valid_identity(&value.incarnation)
532}
533
534fn valid_u64(value: &lodestone_wire::U64) -> bool {
535    value
536        .value
537        .parse::<u64>()
538        .is_ok_and(|parsed| parsed.to_string() == value.value)
539}
540
541fn ordered_successor(predecessor: &lodestone_wire::U64, successor: &lodestone_wire::U64) -> bool {
542    match (predecessor.value.parse::<u64>(), successor.value.parse::<u64>()) {
543        (Ok(predecessor), Ok(successor)) => predecessor.checked_add(1) == Some(successor),
544        _ => false,
545    }
546}
547
548fn revision_at_least(observed: &lodestone_wire::U64, minimum: Option<&lodestone_wire::U64>) -> bool {
549    match (
550        observed.value.parse::<u64>(),
551        minimum.map(|value| value.value.parse::<u64>()),
552    ) {
553        (Ok(_), None) => valid_u64(observed),
554        (Ok(observed), Some(Ok(minimum))) => observed >= minimum,
555        _ => false,
556    }
557}
558
559#[cfg(test)]
560mod tests {
561    use super::*;
562    use lodestone_wire::{ResourceDurabilityClass, ResourceStorageClass, U64};
563
564    fn id(value: char) -> Identity {
565        Identity {
566            value: value.to_string().repeat(32),
567        }
568    }
569
570    fn digest(value: char) -> Digest {
571        Digest {
572            value: value.to_string().repeat(64),
573        }
574    }
575
576    fn locator(value: &str) -> Locator {
577        Locator {
578            authority: "catalog".to_owned(),
579            value: value.to_owned(),
580        }
581    }
582
583    fn resource() -> Resource {
584        Resource {
585            kind: "index".to_owned(),
586            name: "events".to_owned(),
587            incarnation: id('a'),
588        }
589    }
590
591    fn resource_spec() -> IndexResourceSpec {
592        IndexResourceSpec {
593            format_version: u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION),
594            revision: U64 { value: "1".to_owned() },
595            initial_shards: 1,
596            target_shard_bytes: U64 { value: "1".to_owned() },
597            durability: ResourceDurabilityClass::RegionalObject,
598            storage_class: ResourceStorageClass::Standard,
599            ram_admission_weight: 1,
600            ram_hard_max_bytes: U64 { value: "1".to_owned() },
601            nvme_admission_weight: 1,
602            nvme_hard_max_bytes: U64 { value: "1".to_owned() },
603            refresh_objective_millis: 1,
604            memtable_hard_max_bytes: U64 { value: "1".to_owned() },
605            seal_parallelism: 1,
606            merge_parallelism: 1,
607            vector: lodestone_wire::VectorResourcePolicy {
608                quality: lodestone_wire::VectorQualityObjective::Balanced,
609                indexing: lodestone_wire::VectorIndexingObjective::Balanced,
610                residency: None,
611                calibration: None,
612            },
613            lifecycle_binding: None,
614            encryption_required: true,
615            resolver_capability: None,
616        }
617    }
618
619    fn valid_chain() -> Vec<ProvisionRetainedAppendTransition> {
620        let resource = resource();
621        vec![
622            ProvisionRetainedAppendTransition::PreflightVerified,
623            ProvisionRetainedAppendTransition::CatalogPrepared {
624                locator: locator("catalog-1"),
625                semantic_digest: digest('b'),
626                expires_at: U64 { value: "10".to_owned() },
627                resource_spec: resource_spec(),
628            },
629            ProvisionRetainedAppendTransition::CatalogApplied {
630                locator: locator("catalog-1"),
631                resource: resource.clone(),
632            },
633            ProvisionRetainedAppendTransition::IndexActive {
634                resource: resource.clone(),
635                index_revision: U64 { value: "1".to_owned() },
636            },
637            ProvisionRetainedAppendTransition::RolePrepared {
638                locator: locator("role-1"),
639                semantic_digest: digest('c'),
640                expires_at: U64 { value: "20".to_owned() },
641                predecessor_role_version: U64 { value: "4".to_owned() },
642                successor_role_version: U64 { value: "5".to_owned() },
643            },
644            ProvisionRetainedAppendTransition::RoleApplied {
645                locator: locator("role-1"),
646                resolution: RoleAppendApplyResolution::Resolver,
647                target_policy_revision: U64 { value: "9".to_owned() },
648            },
649            ProvisionRetainedAppendTransition::RoleVerified {
650                resource: resource.clone(),
651                role_version: U64 { value: "6".to_owned() },
652                target_policy_revision: Some(U64 { value: "9".to_owned() }),
653            },
654            ProvisionRetainedAppendTransition::DiscoveryVerified {
655                resource: resource.clone(),
656                observed_policy_revision: U64 { value: "10".to_owned() },
657            },
658            ProvisionRetainedAppendTransition::Completed {
659                resource,
660                role_version: U64 { value: "6".to_owned() },
661                observed_policy_revision: U64 { value: "10".to_owned() },
662            },
663        ]
664    }
665
666    #[test]
667    fn reducer_accepts_monotonic_completed_chain() -> Result<(), WorkflowJournalError> {
668        let state = ProvisionRetainedAppendState::replay(&valid_chain())?;
669        assert!(state.completed());
670        assert_eq!(state.resource(), Some(&resource()));
671        Ok(())
672    }
673
674    #[test]
675    fn current_workflow_adopts_exact_catalog_state_without_an_action_locator() -> Result<(), WorkflowJournalError> {
676        let resource = resource();
677        let specification = resource_spec();
678        let transitions = vec![
679            ProvisionRetainedAppendTransition::PreflightVerified,
680            ProvisionRetainedAppendTransition::CatalogAdopted {
681                resource: resource.clone(),
682                resource_spec: specification.clone(),
683            },
684            ProvisionRetainedAppendTransition::IndexActive {
685                resource: resource.clone(),
686                index_revision: U64 { value: "3".to_owned() },
687            },
688            ProvisionRetainedAppendTransition::RoleVerified {
689                resource: resource.clone(),
690                role_version: U64 { value: "6".to_owned() },
691                target_policy_revision: None,
692            },
693            ProvisionRetainedAppendTransition::DiscoveryVerified {
694                resource,
695                observed_policy_revision: U64 { value: "10".to_owned() },
696            },
697        ];
698        let state = ProvisionRetainedAppendState::replay(&transitions)?;
699        assert_eq!(state.resource_spec(), Some(&specification));
700        assert!(state.index_active());
701        Ok(())
702    }
703
704    #[test]
705    fn non_current_workflow_intent_is_rejected() {
706        let intent = ProvisionRetainedAppendIntent {
707            format_version: 1,
708            cluster_id: id('a'),
709            collection_id: id('b'),
710            admin_profile_sha256: [1; 32],
711            consumer_profile_sha256: [2; 32],
712            trust_sha256: [3; 32],
713            definition_sha256: [4; 32],
714            index_name: "events".to_owned(),
715            role_name: "consumer".to_owned(),
716        };
717        assert_eq!(
718            WorkflowJournalIntent::validate(&intent),
719            Err(WorkflowJournalError::InvalidIntent)
720        );
721    }
722
723    #[test]
724    fn intent_and_transitions_reject_zero_authority_values() {
725        let intent = ProvisionRetainedAppendIntent {
726            format_version: PROVISION_RETAINED_APPEND_FORMAT_VERSION,
727            cluster_id: id('0'),
728            collection_id: id('a'),
729            admin_profile_sha256: [1; 32],
730            consumer_profile_sha256: [2; 32],
731            trust_sha256: [3; 32],
732            definition_sha256: [4; 32],
733            index_name: "events".to_owned(),
734            role_name: "consumer".to_owned(),
735        };
736        assert_eq!(intent.validate(), Err(WorkflowJournalError::InvalidIntent));
737
738        let mut intent = intent;
739        intent.cluster_id = id('a');
740        intent.definition_sha256 = [0; 32];
741        assert_eq!(intent.validate(), Err(WorkflowJournalError::InvalidIntent));
742
743        let transition = ProvisionRetainedAppendTransition::CatalogPrepared {
744            locator: locator("catalog-1"),
745            semantic_digest: digest('0'),
746            expires_at: U64 { value: "10".to_owned() },
747            resource_spec: resource_spec(),
748        };
749        assert_eq!(transition.validate(), Err(WorkflowJournalError::InvalidTransition));
750    }
751
752    #[test]
753    fn reducer_rejects_out_of_order_or_post_terminal_records() {
754        assert_eq!(
755            ProvisionRetainedAppendState::replay(&[
756                ProvisionRetainedAppendTransition::PreflightVerified,
757                ProvisionRetainedAppendTransition::IndexActive {
758                    resource: resource(),
759                    index_revision: U64 { value: "1".to_owned() },
760                }
761            ]),
762            Err(WorkflowJournalError::Corrupt)
763        );
764        let mut transitions = valid_chain();
765        transitions.push(ProvisionRetainedAppendTransition::PreflightVerified);
766        assert_eq!(
767            ProvisionRetainedAppendState::replay(&transitions),
768            Err(WorkflowJournalError::Corrupt)
769        );
770    }
771
772    #[test]
773    fn reducer_retires_only_the_exact_role_locator() {
774        let mut transitions = valid_chain();
775        transitions.truncate(5);
776        transitions.push(ProvisionRetainedAppendTransition::RoleAttemptClosed {
777            locator: locator("other"),
778            closure: RoleAppendAttemptClosure::ExpiredUnknown,
779        });
780        assert_eq!(
781            ProvisionRetainedAppendState::replay(&transitions),
782            Err(WorkflowJournalError::Corrupt)
783        );
784    }
785
786    #[test]
787    fn reducer_accepts_a_later_role_version_that_retains_the_scope() -> Result<(), WorkflowJournalError> {
788        let mut transitions = valid_chain();
789        let Some(ProvisionRetainedAppendTransition::Completed { role_version, .. }) = transitions.last_mut() else {
790            return Err(WorkflowJournalError::Corrupt);
791        };
792        *role_version = U64 { value: "7".to_owned() };
793        let state = ProvisionRetainedAppendState::replay(&transitions)?;
794        assert_eq!(state.role_version(), Some(&U64 { value: "7".to_owned() }));
795        Ok(())
796    }
797}
lodestone_client/provisioning.rs

lodestone_client/
provisioning.rs
