Skip to main content

lodestone_client/
development_workflows.rs

1//! Closed recovery models for supported local development workflows.
2
3use std::collections::{BTreeMap, BTreeSet};
4use std::path::{Component, Path, PathBuf};
5
6use lodestone_wire::{
7    Digest, Identity, IndexResourceSpec, LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION, Locator, Resource,
8    ServiceCredentialReservation, U64,
9};
10use serde::{Deserialize, Serialize};
11
12use crate::{
13    LocalOperatorWorkflowKind, PublishedFileIdentity, WorkflowJournalError, WorkflowJournalIntent,
14    WorkflowJournalTransition,
15};
16
17pub const DEVELOPMENT_MATERIALIZE_FORMAT_VERSION: u16 = 1;
18pub const DEVELOPMENT_PKI_SUCCESSOR_FORMAT_VERSION: u16 = 1;
19pub const DEVELOPMENT_RECONCILE_FORMAT_VERSION: u16 = 1;
20const MAX_ENDPOINT_BYTES: usize = 2_048;
21const MAX_NAME_BYTES: usize = 255;
22
23#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
24#[serde(deny_unknown_fields)]
25pub struct DevelopmentMaterializeIntent {
26    pub format_version: u16,
27    pub output_root: String,
28    pub supply_root: String,
29    pub supply_package_sha256: [u8; 32],
30    pub supply_source_revision: String,
31    pub native_api_sha256: [u8; 32],
32    pub native_endpoint: String,
33    pub subscription_endpoint: String,
34    pub internal_endpoint: String,
35    pub bootstrap_endpoint: String,
36}
37
38impl crate::workflow_journal::sealed::Sealed for DevelopmentMaterializeIntent {}
39
40impl WorkflowJournalIntent for DevelopmentMaterializeIntent {
41    type Transition = DevelopmentMaterializeTransition;
42
43    const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::DevelopmentMaterialize;
44
45    fn validate(&self) -> Result<(), WorkflowJournalError> {
46        if self.format_version != DEVELOPMENT_MATERIALIZE_FORMAT_VERSION
47            || !valid_absolute_path(&self.output_root)
48            || !valid_absolute_path(&self.supply_root)
49            || self.output_root == self.supply_root
50            || !valid_sha256(&self.supply_package_sha256)
51            || !valid_git_object(&self.supply_source_revision)
52            || !valid_sha256(&self.native_api_sha256)
53            || !valid_development_endpoints(
54                &self.native_endpoint,
55                &self.subscription_endpoint,
56                &self.internal_endpoint,
57                &self.bootstrap_endpoint,
58            )
59        {
60            return Err(WorkflowJournalError::InvalidIntent);
61        }
62        Ok(())
63    }
64}
65
66#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
67#[serde(rename_all = "snake_case")]
68pub enum DevelopmentMaterializeArtifact {
69    CaCertificate,
70    CaPrivateKey,
71    ProcessCertificate,
72    ProcessPrivateKey,
73    GatewayCertificate,
74    GatewayPrivateKey,
75    BootstrapCredential,
76    BootstrapIntent,
77    ServerConfiguration,
78    DevelopmentIndexDefinition,
79    InstanceManifest,
80}
81
82#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
83#[serde(rename_all = "snake_case")]
84pub enum DevelopmentMaterializePrivatePlan {
85    Pki,
86}
87
88impl DevelopmentMaterializeArtifact {
89    fn is_manifest(self) -> bool {
90        self == Self::InstanceManifest
91    }
92
93    fn required_before_manifest() -> BTreeSet<Self> {
94        [
95            Self::CaCertificate,
96            Self::CaPrivateKey,
97            Self::ProcessCertificate,
98            Self::ProcessPrivateKey,
99            Self::GatewayCertificate,
100            Self::GatewayPrivateKey,
101            Self::BootstrapCredential,
102            Self::BootstrapIntent,
103            Self::ServerConfiguration,
104            Self::DevelopmentIndexDefinition,
105        ]
106        .into_iter()
107        .collect()
108    }
109}
110
111#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
112#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
113pub enum DevelopmentMaterializeTransition {
114    InstanceIdentityRecorded {
115        instance_identity: String,
116    },
117    ClockRecorded {
118        unix_millis: U64,
119    },
120    PrivatePlanStaged {
121        plan: DevelopmentMaterializePrivatePlan,
122        file: PublishedFileIdentity,
123    },
124    PrivatePlanPublished {
125        plan: DevelopmentMaterializePrivatePlan,
126        file: PublishedFileIdentity,
127    },
128    ArtifactStaged {
129        artifact: DevelopmentMaterializeArtifact,
130        file: PublishedFileIdentity,
131    },
132    ArtifactPublished {
133        artifact: DevelopmentMaterializeArtifact,
134        file: PublishedFileIdentity,
135        #[serde(default, skip_serializing_if = "Option::is_none")]
136        public_sha256: Option<[u8; 32]>,
137    },
138    Verified {
139        manifest_sha256: [u8; 32],
140    },
141    Completed,
142}
143
144impl crate::workflow_journal::sealed::Sealed for DevelopmentMaterializeTransition {}
145
146impl WorkflowJournalTransition for DevelopmentMaterializeTransition {
147    fn validate(&self) -> Result<(), WorkflowJournalError> {
148        let valid = match self {
149            Self::InstanceIdentityRecorded { instance_identity } => valid_instance_identity(instance_identity),
150            Self::ClockRecorded { unix_millis } => valid_u64(unix_millis),
151            Self::PrivatePlanStaged { file, .. }
152            | Self::PrivatePlanPublished { file, .. }
153            | Self::ArtifactStaged { file, .. } => file.is_valid(),
154            Self::ArtifactPublished {
155                artifact,
156                file,
157                public_sha256,
158            } => {
159                file.is_valid()
160                    && public_sha256.as_ref().is_none_or(valid_sha256)
161                    && (!artifact.is_manifest() || public_sha256.as_ref().is_some_and(valid_sha256))
162            }
163            Self::Verified { manifest_sha256 } => valid_sha256(manifest_sha256),
164            Self::Completed => true,
165        };
166        if valid {
167            Ok(())
168        } else {
169            Err(WorkflowJournalError::InvalidTransition)
170        }
171    }
172}
173
174#[derive(Clone, Debug, Default, PartialEq)]
175pub struct DevelopmentMaterializeState {
176    instance_identity: Option<String>,
177    unix_millis: Option<U64>,
178    private_plans: BTreeMap<DevelopmentMaterializePrivatePlan, PublishedFileIdentity>,
179    published_private_plans: BTreeMap<DevelopmentMaterializePrivatePlan, PublishedFileIdentity>,
180    staged: BTreeMap<DevelopmentMaterializeArtifact, PublishedFileIdentity>,
181    published: BTreeMap<DevelopmentMaterializeArtifact, PublishedFileIdentity>,
182    published_manifest_sha256: Option<[u8; 32]>,
183    manifest_sha256: Option<[u8; 32]>,
184    completed: bool,
185}
186
187impl DevelopmentMaterializeState {
188    pub fn replay(transitions: &[DevelopmentMaterializeTransition]) -> Result<Self, WorkflowJournalError> {
189        let mut state = Self::default();
190        for transition in transitions {
191            transition.validate()?;
192            state.apply(transition)?;
193        }
194        Ok(state)
195    }
196
197    fn apply(&mut self, transition: &DevelopmentMaterializeTransition) -> Result<(), WorkflowJournalError> {
198        if self.completed {
199            return Err(WorkflowJournalError::Corrupt);
200        }
201        match transition {
202            DevelopmentMaterializeTransition::InstanceIdentityRecorded { instance_identity }
203                if self.instance_identity.is_none() && self.unix_millis.is_none() =>
204            {
205                self.instance_identity = Some(instance_identity.clone());
206            }
207            DevelopmentMaterializeTransition::ClockRecorded { unix_millis }
208                if self.instance_identity.is_some() && self.unix_millis.is_none() =>
209            {
210                self.unix_millis = Some(unix_millis.clone());
211            }
212            DevelopmentMaterializeTransition::PrivatePlanStaged { plan, file }
213                if self.unix_millis.is_some()
214                    && self.staged.is_empty()
215                    && self.published.is_empty()
216                    && !self.private_plans.contains_key(plan) =>
217            {
218                self.private_plans.insert(*plan, file.clone());
219            }
220            DevelopmentMaterializeTransition::PrivatePlanPublished { plan, file }
221                if !self.published_private_plans.contains_key(plan) && self.private_plans.get(plan) == Some(file) =>
222            {
223                self.published_private_plans.insert(*plan, file.clone());
224            }
225            DevelopmentMaterializeTransition::ArtifactStaged { artifact, file }
226                if self
227                    .published_private_plans
228                    .contains_key(&DevelopmentMaterializePrivatePlan::Pki)
229                    && !self.staged.contains_key(artifact)
230                    && !self.published.contains_key(artifact)
231                    && (!artifact.is_manifest()
232                        || self.published.keys().copied().collect::<BTreeSet<_>>()
233                            == DevelopmentMaterializeArtifact::required_before_manifest()) =>
234            {
235                self.staged.insert(*artifact, file.clone());
236            }
237            DevelopmentMaterializeTransition::ArtifactPublished {
238                artifact,
239                file,
240                public_sha256,
241            } if !self.published.contains_key(artifact) && self.staged.get(artifact) == Some(file) => {
242                self.published.insert(*artifact, file.clone());
243                if artifact.is_manifest() {
244                    self.published_manifest_sha256 = *public_sha256;
245                }
246            }
247            DevelopmentMaterializeTransition::Verified { manifest_sha256 }
248                if self
249                    .published
250                    .contains_key(&DevelopmentMaterializeArtifact::InstanceManifest)
251                    && self.published_manifest_sha256.as_ref() == Some(manifest_sha256)
252                    && self.manifest_sha256.is_none() =>
253            {
254                self.manifest_sha256 = Some(*manifest_sha256);
255            }
256            DevelopmentMaterializeTransition::Completed if self.manifest_sha256.is_some() => {
257                self.completed = true;
258            }
259            _ => return Err(WorkflowJournalError::Corrupt),
260        }
261        Ok(())
262    }
263
264    #[must_use]
265    pub fn instance_identity(&self) -> Option<&str> {
266        self.instance_identity.as_deref()
267    }
268
269    #[must_use]
270    pub const fn clock(&self) -> Option<&U64> {
271        self.unix_millis.as_ref()
272    }
273
274    #[must_use]
275    pub fn staged(&self, artifact: DevelopmentMaterializeArtifact) -> Option<&PublishedFileIdentity> {
276        self.staged.get(&artifact)
277    }
278
279    #[must_use]
280    pub fn private_plan(&self, plan: DevelopmentMaterializePrivatePlan) -> Option<&PublishedFileIdentity> {
281        self.private_plans.get(&plan)
282    }
283
284    #[must_use]
285    pub fn published_private_plan(&self, plan: DevelopmentMaterializePrivatePlan) -> Option<&PublishedFileIdentity> {
286        self.published_private_plans.get(&plan)
287    }
288
289    #[must_use]
290    pub fn published(&self, artifact: DevelopmentMaterializeArtifact) -> Option<&PublishedFileIdentity> {
291        self.published.get(&artifact)
292    }
293
294    #[must_use]
295    pub const fn completed(&self) -> bool {
296        self.completed
297    }
298}
299
300#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
301#[serde(rename_all = "snake_case")]
302pub enum DevelopmentPkiSuccessorOperation {
303    RenewLeaves,
304    RotateCa,
305}
306
307#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
308#[serde(deny_unknown_fields)]
309pub struct DevelopmentPkiSuccessorIntent {
310    pub format_version: u16,
311    pub operation: DevelopmentPkiSuccessorOperation,
312    pub instance_identity: String,
313    pub parent_manifest_path: String,
314    pub parent_manifest_sha256: [u8; 32],
315    pub successor_root: String,
316    pub parent_instance_generation: U64,
317    pub successor_instance_generation: U64,
318    pub parent_ca_generation: U64,
319    pub successor_ca_generation: U64,
320    pub parent_process_leaf_generation: U64,
321    pub successor_process_leaf_generation: U64,
322    pub parent_gateway_leaf_generation: U64,
323    pub successor_gateway_leaf_generation: U64,
324}
325
326impl crate::workflow_journal::sealed::Sealed for DevelopmentPkiSuccessorIntent {}
327
328impl WorkflowJournalIntent for DevelopmentPkiSuccessorIntent {
329    type Transition = DevelopmentPkiSuccessorTransition;
330
331    const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::DevelopmentPkiSuccessor;
332
333    fn validate(&self) -> Result<(), WorkflowJournalError> {
334        let instance_generations =
335            ordered_successor(&self.parent_instance_generation, &self.successor_instance_generation);
336        let process_generations = ordered_successor(
337            &self.parent_process_leaf_generation,
338            &self.successor_process_leaf_generation,
339        );
340        let gateway_generations = ordered_successor(
341            &self.parent_gateway_leaf_generation,
342            &self.successor_gateway_leaf_generation,
343        );
344        let ca_generations = match self.operation {
345            DevelopmentPkiSuccessorOperation::RenewLeaves => {
346                canonical_u64(&self.parent_ca_generation).filter(|value| *value > 0)
347                    == canonical_u64(&self.successor_ca_generation).filter(|value| *value > 0)
348            }
349            DevelopmentPkiSuccessorOperation::RotateCa => {
350                ordered_successor(&self.parent_ca_generation, &self.successor_ca_generation)
351            }
352        };
353        let parent_manifest = Path::new(&self.parent_manifest_path);
354        let parent_root = parent_manifest.parent();
355        if self.format_version != DEVELOPMENT_PKI_SUCCESSOR_FORMAT_VERSION
356            || !valid_instance_identity(&self.instance_identity)
357            || !valid_absolute_path(&self.parent_manifest_path)
358            || !valid_sha256(&self.parent_manifest_sha256)
359            || !valid_absolute_path(&self.successor_root)
360            || parent_manifest.file_name().and_then(|name| name.to_str()) != Some("instance.json")
361            || parent_root.is_none_or(|root| paths_overlap(root, Path::new(&self.successor_root)))
362            || !instance_generations
363            || !process_generations
364            || !gateway_generations
365            || !ca_generations
366        {
367            return Err(WorkflowJournalError::InvalidIntent);
368        }
369        Ok(())
370    }
371}
372
373fn paths_overlap(left: &Path, right: &Path) -> bool {
374    left.starts_with(right) || right.starts_with(left)
375}
376
377#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
378#[serde(rename_all = "snake_case")]
379pub enum DevelopmentPkiSuccessorArtifact {
380    CaCertificate,
381    CaPrivateKey,
382    ProcessCertificate,
383    ProcessPrivateKey,
384    GatewayCertificate,
385    GatewayPrivateKey,
386    ServerConfiguration,
387    InstanceManifest,
388}
389
390#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
391#[serde(rename_all = "snake_case")]
392pub enum DevelopmentPkiSuccessorPrivatePlan {
393    Pki,
394}
395
396#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
397#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
398pub enum DevelopmentPkiSuccessorTransition {
399    ClockRecorded {
400        unix_millis: U64,
401    },
402    PrivatePlanStaged {
403        plan: DevelopmentPkiSuccessorPrivatePlan,
404        file: PublishedFileIdentity,
405    },
406    PrivatePlanPublished {
407        plan: DevelopmentPkiSuccessorPrivatePlan,
408        file: PublishedFileIdentity,
409    },
410    ArtifactStaged {
411        artifact: DevelopmentPkiSuccessorArtifact,
412        file: PublishedFileIdentity,
413    },
414    ArtifactPublished {
415        artifact: DevelopmentPkiSuccessorArtifact,
416        file: PublishedFileIdentity,
417        #[serde(default, skip_serializing_if = "Option::is_none")]
418        public_sha256: Option<[u8; 32]>,
419    },
420    Verified {
421        manifest_sha256: [u8; 32],
422    },
423    Completed,
424}
425
426impl crate::workflow_journal::sealed::Sealed for DevelopmentPkiSuccessorTransition {}
427
428impl WorkflowJournalTransition for DevelopmentPkiSuccessorTransition {
429    fn validate(&self) -> Result<(), WorkflowJournalError> {
430        let valid = match self {
431            Self::ClockRecorded { unix_millis } => valid_u64(unix_millis),
432            Self::PrivatePlanStaged { file, .. }
433            | Self::PrivatePlanPublished { file, .. }
434            | Self::ArtifactStaged { file, .. } => file.is_valid(),
435            Self::ArtifactPublished {
436                artifact,
437                file,
438                public_sha256,
439            } => {
440                file.is_valid()
441                    && public_sha256.as_ref().is_none_or(valid_sha256)
442                    && (*artifact != DevelopmentPkiSuccessorArtifact::InstanceManifest
443                        || public_sha256.as_ref().is_some_and(valid_sha256))
444            }
445            Self::Verified { manifest_sha256 } => valid_sha256(manifest_sha256),
446            Self::Completed => true,
447        };
448        if valid {
449            Ok(())
450        } else {
451            Err(WorkflowJournalError::InvalidTransition)
452        }
453    }
454}
455
456#[derive(Clone, Debug, Default, PartialEq)]
457pub struct DevelopmentPkiSuccessorState {
458    clock: Option<U64>,
459    private_plans: BTreeMap<DevelopmentPkiSuccessorPrivatePlan, PublishedFileIdentity>,
460    published_private_plans: BTreeMap<DevelopmentPkiSuccessorPrivatePlan, PublishedFileIdentity>,
461    staged: BTreeMap<DevelopmentPkiSuccessorArtifact, PublishedFileIdentity>,
462    published: BTreeMap<DevelopmentPkiSuccessorArtifact, PublishedFileIdentity>,
463    published_manifest_sha256: Option<[u8; 32]>,
464    manifest_sha256: Option<[u8; 32]>,
465    completed: bool,
466}
467
468impl DevelopmentPkiSuccessorState {
469    pub fn replay(
470        intent: &DevelopmentPkiSuccessorIntent,
471        transitions: &[DevelopmentPkiSuccessorTransition],
472    ) -> Result<Self, WorkflowJournalError> {
473        intent.validate()?;
474        let mut state = Self::default();
475        for transition in transitions {
476            transition.validate()?;
477            state.apply(intent.operation, transition)?;
478        }
479        Ok(state)
480    }
481
482    fn apply(
483        &mut self,
484        operation: DevelopmentPkiSuccessorOperation,
485        transition: &DevelopmentPkiSuccessorTransition,
486    ) -> Result<(), WorkflowJournalError> {
487        if self.completed {
488            return Err(WorkflowJournalError::Corrupt);
489        }
490        match transition {
491            DevelopmentPkiSuccessorTransition::ClockRecorded { unix_millis } if self.clock.is_none() => {
492                self.clock = Some(unix_millis.clone());
493            }
494            DevelopmentPkiSuccessorTransition::PrivatePlanStaged { plan, file }
495                if self.clock.is_some()
496                    && self.staged.is_empty()
497                    && self.published.is_empty()
498                    && !self.private_plans.contains_key(plan) =>
499            {
500                self.private_plans.insert(*plan, file.clone());
501            }
502            DevelopmentPkiSuccessorTransition::PrivatePlanPublished { plan, file }
503                if !self.published_private_plans.contains_key(plan) && self.private_plans.get(plan) == Some(file) =>
504            {
505                self.published_private_plans.insert(*plan, file.clone());
506            }
507            DevelopmentPkiSuccessorTransition::ArtifactStaged { artifact, file }
508                if self.clock.is_some()
509                    && self
510                        .published_private_plans
511                        .contains_key(&DevelopmentPkiSuccessorPrivatePlan::Pki)
512                    && !self.staged.contains_key(artifact)
513                    && !self.published.contains_key(artifact)
514                    && (artifact_admitted_for_pki_operation(operation, *artifact)
515                        || (*artifact == DevelopmentPkiSuccessorArtifact::InstanceManifest
516                            && self.published.keys().copied().collect::<BTreeSet<_>>()
517                                == required_pki_artifacts(operation))) =>
518            {
519                self.staged.insert(*artifact, file.clone());
520            }
521            DevelopmentPkiSuccessorTransition::ArtifactPublished {
522                artifact,
523                file,
524                public_sha256,
525            } if !self.published.contains_key(artifact) && self.staged.get(artifact) == Some(file) => {
526                self.published.insert(*artifact, file.clone());
527                if *artifact == DevelopmentPkiSuccessorArtifact::InstanceManifest {
528                    self.published_manifest_sha256 = *public_sha256;
529                }
530            }
531            DevelopmentPkiSuccessorTransition::Verified { manifest_sha256 }
532                if self
533                    .published
534                    .contains_key(&DevelopmentPkiSuccessorArtifact::InstanceManifest)
535                    && self.published_manifest_sha256.as_ref() == Some(manifest_sha256)
536                    && self.manifest_sha256.is_none() =>
537            {
538                self.manifest_sha256 = Some(*manifest_sha256);
539            }
540            DevelopmentPkiSuccessorTransition::Completed if self.manifest_sha256.is_some() => {
541                self.completed = true;
542            }
543            _ => return Err(WorkflowJournalError::Corrupt),
544        }
545        Ok(())
546    }
547
548    #[must_use]
549    pub const fn clock(&self) -> Option<&U64> {
550        self.clock.as_ref()
551    }
552
553    #[must_use]
554    pub fn private_plan(&self, plan: DevelopmentPkiSuccessorPrivatePlan) -> Option<&PublishedFileIdentity> {
555        self.private_plans.get(&plan)
556    }
557
558    #[must_use]
559    pub fn published_private_plan(&self, plan: DevelopmentPkiSuccessorPrivatePlan) -> Option<&PublishedFileIdentity> {
560        self.published_private_plans.get(&plan)
561    }
562
563    #[must_use]
564    pub fn staged(&self, artifact: DevelopmentPkiSuccessorArtifact) -> Option<&PublishedFileIdentity> {
565        self.staged.get(&artifact)
566    }
567
568    #[must_use]
569    pub fn published(&self, artifact: DevelopmentPkiSuccessorArtifact) -> Option<&PublishedFileIdentity> {
570        self.published.get(&artifact)
571    }
572
573    #[must_use]
574    pub const fn completed(&self) -> bool {
575        self.completed
576    }
577}
578
579fn artifact_admitted_for_pki_operation(
580    operation: DevelopmentPkiSuccessorOperation,
581    artifact: DevelopmentPkiSuccessorArtifact,
582) -> bool {
583    artifact != DevelopmentPkiSuccessorArtifact::InstanceManifest
584        && (operation == DevelopmentPkiSuccessorOperation::RotateCa
585            || !matches!(
586                artifact,
587                DevelopmentPkiSuccessorArtifact::CaCertificate | DevelopmentPkiSuccessorArtifact::CaPrivateKey
588            ))
589}
590
591fn required_pki_artifacts(operation: DevelopmentPkiSuccessorOperation) -> BTreeSet<DevelopmentPkiSuccessorArtifact> {
592    let mut required = [
593        DevelopmentPkiSuccessorArtifact::ProcessCertificate,
594        DevelopmentPkiSuccessorArtifact::ProcessPrivateKey,
595        DevelopmentPkiSuccessorArtifact::GatewayCertificate,
596        DevelopmentPkiSuccessorArtifact::GatewayPrivateKey,
597        DevelopmentPkiSuccessorArtifact::ServerConfiguration,
598    ]
599    .into_iter()
600    .collect::<BTreeSet<_>>();
601    if operation == DevelopmentPkiSuccessorOperation::RotateCa {
602        required.insert(DevelopmentPkiSuccessorArtifact::CaCertificate);
603        required.insert(DevelopmentPkiSuccessorArtifact::CaPrivateKey);
604    }
605    required
606}
607
608#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
609#[serde(deny_unknown_fields)]
610pub struct DevelopmentReconcileIntent {
611    pub format_version: u16,
612    pub instance_root: String,
613    pub instance_manifest_sha256: [u8; 32],
614    pub index_definition_path: String,
615    pub index_definition_sha256: [u8; 32],
616    pub credential_output: String,
617    pub profile_output: String,
618    pub material_output: String,
619    pub timeout_millis: U64,
620    pub reservation_ttl_millis: U64,
621    pub credential_ttl_millis: U64,
622}
623
624impl crate::workflow_journal::sealed::Sealed for DevelopmentReconcileIntent {}
625
626impl WorkflowJournalIntent for DevelopmentReconcileIntent {
627    type Transition = DevelopmentReconcileTransition;
628
629    const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::DevelopmentReconcile;
630
631    fn validate(&self) -> Result<(), WorkflowJournalError> {
632        let paths = [
633            &self.instance_root,
634            &self.index_definition_path,
635            &self.credential_output,
636            &self.profile_output,
637            &self.material_output,
638        ];
639        let distinct_outputs = [&self.credential_output, &self.profile_output, &self.material_output]
640            .into_iter()
641            .collect::<BTreeSet<_>>()
642            .len()
643            == 3;
644        if self.format_version != DEVELOPMENT_RECONCILE_FORMAT_VERSION
645            || paths.into_iter().any(|path| !valid_absolute_path(path))
646            || !distinct_outputs
647            || !valid_sha256(&self.instance_manifest_sha256)
648            || !valid_sha256(&self.index_definition_sha256)
649            || !valid_positive_bounded_duration(&self.timeout_millis)
650            || !valid_positive_bounded_duration(&self.reservation_ttl_millis)
651            || !valid_positive_bounded_duration(&self.credential_ttl_millis)
652            || !ordered_u64(&self.reservation_ttl_millis, &self.credential_ttl_millis)
653        {
654            return Err(WorkflowJournalError::InvalidIntent);
655        }
656        Ok(())
657    }
658}
659
660#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
661#[serde(rename_all = "snake_case")]
662pub enum DevelopmentReconcileRoleTarget {
663    Administrator,
664    Consumer,
665}
666
667#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
668#[serde(rename_all = "snake_case")]
669pub enum DevelopmentReconcileTerminalOutcome {
670    EffectDeadlineExpired,
671    Conflict,
672    ReservationExpiredNoEffect,
673}
674
675#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
676#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
677pub enum DevelopmentReconcileTransition {
678    EffectDeadlineRecorded {
679        unix_millis: U64,
680    },
681    AuthorityObserved {
682        cluster_id: Identity,
683        collection_id: Identity,
684        schema_sha256: Digest,
685        logical_time: U64,
686        reservation_expires_at: U64,
687        credential_expires_at: U64,
688    },
689    CatalogPrepared {
690        locator: Locator,
691        semantic_digest: Digest,
692        expires_at: U64,
693        resource_spec: IndexResourceSpec,
694    },
695    CatalogApplied {
696        locator: Locator,
697        resource: Resource,
698    },
699    IndexActive {
700        resource: Resource,
701        index_revision: U64,
702    },
703    RolePrepared {
704        target: DevelopmentReconcileRoleTarget,
705        locator: Locator,
706        semantic_digest: Digest,
707        expires_at: U64,
708        predecessor_role_version: U64,
709        successor_role_version: U64,
710    },
711    RoleApplied {
712        target: DevelopmentReconcileRoleTarget,
713        locator: Locator,
714        target_policy_revision: U64,
715    },
716    RoleVerified {
717        target: DevelopmentReconcileRoleTarget,
718        role_version: U64,
719        observed_policy_revision: U64,
720    },
721    CredentialReserved {
722        reservation: ServiceCredentialReservation,
723    },
724    CredentialStaged {
725        file: PublishedFileIdentity,
726    },
727    CredentialActivated {
728        principal_id: Identity,
729        api_key_id: Identity,
730        generation: U64,
731    },
732    CredentialPublished {
733        file: PublishedFileIdentity,
734    },
735    ProfilePublished {
736        file: PublishedFileIdentity,
737        profile_sha256: [u8; 32],
738    },
739    ConsumerVerified {
740        observed_policy_revision: U64,
741    },
742    MaterialPublished {
743        file: PublishedFileIdentity,
744        material_sha256: [u8; 32],
745    },
746    Terminal {
747        outcome: DevelopmentReconcileTerminalOutcome,
748    },
749    Completed,
750}
751
752impl crate::workflow_journal::sealed::Sealed for DevelopmentReconcileTransition {}
753
754impl WorkflowJournalTransition for DevelopmentReconcileTransition {
755    fn validate(&self) -> Result<(), WorkflowJournalError> {
756        let valid = match self {
757            Self::EffectDeadlineRecorded { unix_millis } => valid_u64(unix_millis),
758            Self::AuthorityObserved {
759                cluster_id,
760                collection_id,
761                schema_sha256,
762                logical_time,
763                reservation_expires_at,
764                credential_expires_at,
765            } => {
766                valid_identity(cluster_id)
767                    && valid_identity(collection_id)
768                    && valid_digest(schema_sha256)
769                    && valid_u64(logical_time)
770                    && ordered_u64(logical_time, reservation_expires_at)
771                    && ordered_u64(reservation_expires_at, credential_expires_at)
772            }
773            Self::CatalogPrepared {
774                locator,
775                semantic_digest,
776                expires_at,
777                resource_spec,
778            } => {
779                valid_locator(locator)
780                    && valid_digest(semantic_digest)
781                    && valid_u64(expires_at)
782                    && resource_spec.format_version == u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION)
783            }
784            Self::CatalogApplied { locator, resource } => valid_locator(locator) && valid_resource(resource),
785            Self::IndexActive {
786                resource,
787                index_revision,
788            } => valid_resource(resource) && valid_u64(index_revision),
789            Self::RolePrepared {
790                locator,
791                semantic_digest,
792                expires_at,
793                predecessor_role_version,
794                successor_role_version,
795                ..
796            } => {
797                valid_locator(locator)
798                    && valid_digest(semantic_digest)
799                    && valid_u64(expires_at)
800                    && ordered_successor(predecessor_role_version, successor_role_version)
801            }
802            Self::RoleApplied {
803                locator,
804                target_policy_revision,
805                ..
806            } => valid_locator(locator) && valid_u64(target_policy_revision),
807            Self::RoleVerified {
808                role_version,
809                observed_policy_revision,
810                ..
811            } => valid_u64(role_version) && valid_u64(observed_policy_revision),
812            Self::CredentialReserved { reservation } => valid_reservation(reservation),
813            Self::CredentialStaged { file } | Self::CredentialPublished { file } => file.is_valid(),
814            Self::CredentialActivated {
815                principal_id,
816                api_key_id,
817                generation,
818            } => valid_identity(principal_id) && valid_identity(api_key_id) && valid_u64(generation),
819            Self::ProfilePublished { file, profile_sha256 } => file.is_valid() && valid_sha256(profile_sha256),
820            Self::ConsumerVerified {
821                observed_policy_revision,
822            } => valid_u64(observed_policy_revision),
823            Self::MaterialPublished { file, material_sha256 } => file.is_valid() && valid_sha256(material_sha256),
824            Self::Terminal { .. } => true,
825            Self::Completed => true,
826        };
827        if valid {
828            Ok(())
829        } else {
830            Err(WorkflowJournalError::InvalidTransition)
831        }
832    }
833}
834
835#[derive(Clone, Debug, Default, PartialEq)]
836pub struct DevelopmentReconcileState {
837    effect_deadline: Option<U64>,
838    binding: Option<(Identity, Identity, Digest, U64, U64, U64)>,
839    catalog_preparation: Option<(Locator, Digest, U64, IndexResourceSpec)>,
840    resource: Option<Resource>,
841    index_active: bool,
842    role_preparations: BTreeMap<DevelopmentReconcileRoleTarget, (Locator, Digest, U64, U64, U64)>,
843    role_applied: BTreeMap<DevelopmentReconcileRoleTarget, U64>,
844    role_verified: BTreeMap<DevelopmentReconcileRoleTarget, (U64, U64)>,
845    reservation: Option<ServiceCredentialReservation>,
846    credential_staged: Option<PublishedFileIdentity>,
847    credential_activated: bool,
848    credential_published: Option<PublishedFileIdentity>,
849    profile_published: Option<PublishedFileIdentity>,
850    consumer_verified: Option<U64>,
851    material_published: Option<PublishedFileIdentity>,
852    terminal: Option<DevelopmentReconcileTerminalOutcome>,
853    completed: bool,
854}
855
856impl DevelopmentReconcileState {
857    pub fn replay(
858        intent: &DevelopmentReconcileIntent,
859        transitions: &[DevelopmentReconcileTransition],
860    ) -> Result<Self, WorkflowJournalError> {
861        intent.validate()?;
862        let mut state = Self::default();
863        for transition in transitions {
864            transition.validate()?;
865            state.apply(intent, transition)?;
866        }
867        Ok(state)
868    }
869
870    fn apply(
871        &mut self,
872        intent: &DevelopmentReconcileIntent,
873        transition: &DevelopmentReconcileTransition,
874    ) -> Result<(), WorkflowJournalError> {
875        if self.completed || self.terminal.is_some() {
876            return Err(WorkflowJournalError::Corrupt);
877        }
878        match transition {
879            DevelopmentReconcileTransition::EffectDeadlineRecorded { unix_millis }
880                if self.effect_deadline.is_none() =>
881            {
882                self.effect_deadline = Some(unix_millis.clone());
883            }
884            DevelopmentReconcileTransition::AuthorityObserved {
885                cluster_id,
886                collection_id,
887                schema_sha256,
888                logical_time,
889                reservation_expires_at,
890                credential_expires_at,
891            } if self.effect_deadline.is_some()
892                && self.binding.is_none()
893                && checked_add_u64(logical_time, &intent.reservation_ttl_millis)
894                    == canonical_u64(reservation_expires_at)
895                && checked_add_u64(logical_time, &intent.credential_ttl_millis)
896                    == canonical_u64(credential_expires_at) =>
897            {
898                self.binding = Some((
899                    cluster_id.clone(),
900                    collection_id.clone(),
901                    schema_sha256.clone(),
902                    logical_time.clone(),
903                    reservation_expires_at.clone(),
904                    credential_expires_at.clone(),
905                ));
906            }
907            DevelopmentReconcileTransition::CatalogPrepared {
908                locator,
909                semantic_digest,
910                expires_at,
911                resource_spec,
912            } if self.binding.is_some() && self.catalog_preparation.is_none() && self.resource.is_none() => {
913                self.catalog_preparation = Some((
914                    locator.clone(),
915                    semantic_digest.clone(),
916                    expires_at.clone(),
917                    resource_spec.clone(),
918                ));
919            }
920            DevelopmentReconcileTransition::CatalogApplied { locator, resource }
921                if self
922                    .catalog_preparation
923                    .as_ref()
924                    .is_some_and(|(expected, _, _, _)| expected == locator)
925                    && self.resource.is_none() =>
926            {
927                self.resource = Some(resource.clone());
928            }
929            DevelopmentReconcileTransition::IndexActive { resource, .. }
930                if self.binding.is_some()
931                    && self.resource.as_ref().is_none_or(|current| current == resource)
932                    && !self.index_active =>
933            {
934                self.resource = Some(resource.clone());
935                self.index_active = true;
936            }
937            DevelopmentReconcileTransition::RolePrepared {
938                target,
939                locator,
940                semantic_digest,
941                expires_at,
942                predecessor_role_version,
943                successor_role_version,
944            } if self.role_target_is_next(*target) && !self.role_preparations.contains_key(target) => {
945                self.role_preparations.insert(
946                    *target,
947                    (
948                        locator.clone(),
949                        semantic_digest.clone(),
950                        expires_at.clone(),
951                        predecessor_role_version.clone(),
952                        successor_role_version.clone(),
953                    ),
954                );
955            }
956            DevelopmentReconcileTransition::RoleApplied {
957                target,
958                locator,
959                target_policy_revision,
960            } if self
961                .role_preparations
962                .get(target)
963                .is_some_and(|(expected, _, _, _, _)| expected == locator)
964                && !self.role_applied.contains_key(target) =>
965            {
966                self.role_applied.insert(*target, target_policy_revision.clone());
967            }
968            DevelopmentReconcileTransition::RoleVerified {
969                target,
970                role_version,
971                observed_policy_revision,
972            } if self.role_target_is_next(*target)
973                && self.role_verification_matches(*target, role_version, observed_policy_revision) =>
974            {
975                self.role_verified
976                    .insert(*target, (role_version.clone(), observed_policy_revision.clone()));
977            }
978            DevelopmentReconcileTransition::CredentialReserved { reservation }
979                if self
980                    .role_verified
981                    .contains_key(&DevelopmentReconcileRoleTarget::Consumer)
982                    && self.reservation.is_none() =>
983            {
984                let Some((_, _, _, _, reservation_expires_at, credential_expires_at)) = self.binding.as_ref() else {
985                    return Err(WorkflowJournalError::Corrupt);
986                };
987                if &reservation.reservation_expires_at != reservation_expires_at
988                    || &reservation.credential_expires_at != credential_expires_at
989                {
990                    return Err(WorkflowJournalError::Corrupt);
991                }
992                self.reservation = Some(reservation.clone());
993            }
994            DevelopmentReconcileTransition::CredentialStaged { file }
995                if self.reservation.is_some() && self.credential_staged.is_none() =>
996            {
997                self.credential_staged = Some(file.clone());
998            }
999            DevelopmentReconcileTransition::CredentialActivated {
1000                principal_id,
1001                api_key_id,
1002                generation,
1003            } if self.reservation.as_ref().is_some_and(|reservation| {
1004                &reservation.principal_id == principal_id
1005                    && &reservation.api_key_id == api_key_id
1006                    && &reservation.generation == generation
1007            }) && self.credential_staged.is_some()
1008                && !self.credential_activated =>
1009            {
1010                self.credential_activated = true;
1011            }
1012            DevelopmentReconcileTransition::CredentialPublished { file }
1013                if self.credential_activated
1014                    && self.credential_staged.as_ref() == Some(file)
1015                    && self.credential_published.is_none() =>
1016            {
1017                self.credential_published = Some(file.clone());
1018            }
1019            DevelopmentReconcileTransition::ProfilePublished { file, .. }
1020                if self.credential_published.is_some() && self.profile_published.is_none() =>
1021            {
1022                self.profile_published = Some(file.clone());
1023            }
1024            DevelopmentReconcileTransition::ConsumerVerified {
1025                observed_policy_revision,
1026            } if self.profile_published.is_some() && self.consumer_verified.is_none() => {
1027                self.consumer_verified = Some(observed_policy_revision.clone());
1028            }
1029            DevelopmentReconcileTransition::MaterialPublished { file, .. }
1030                if self.consumer_verified.is_some() && self.material_published.is_none() =>
1031            {
1032                self.material_published = Some(file.clone());
1033            }
1034            DevelopmentReconcileTransition::Terminal { outcome, .. } if self.effect_deadline.is_some() => {
1035                self.terminal = Some(*outcome);
1036            }
1037            DevelopmentReconcileTransition::Completed if self.material_published.is_some() => {
1038                self.completed = true;
1039            }
1040            _ => return Err(WorkflowJournalError::Corrupt),
1041        }
1042        Ok(())
1043    }
1044
1045    fn role_target_is_next(&self, target: DevelopmentReconcileRoleTarget) -> bool {
1046        self.index_active
1047            && match target {
1048                DevelopmentReconcileRoleTarget::Administrator => !self
1049                    .role_verified
1050                    .contains_key(&DevelopmentReconcileRoleTarget::Administrator),
1051                DevelopmentReconcileRoleTarget::Consumer => {
1052                    self.role_verified
1053                        .contains_key(&DevelopmentReconcileRoleTarget::Administrator)
1054                        && !self
1055                            .role_verified
1056                            .contains_key(&DevelopmentReconcileRoleTarget::Consumer)
1057                }
1058            }
1059    }
1060
1061    fn role_verification_matches(
1062        &self,
1063        target: DevelopmentReconcileRoleTarget,
1064        role_version: &U64,
1065        observed_policy_revision: &U64,
1066    ) -> bool {
1067        match self.role_preparations.get(&target) {
1068            Some((_, _, _, _, successor_role_version)) => {
1069                successor_role_version == role_version
1070                    && self.role_applied.get(&target).is_some_and(|target_policy_revision| {
1071                        ordered_or_equal_u64(target_policy_revision, observed_policy_revision)
1072                    })
1073            }
1074            None => !self.role_applied.contains_key(&target),
1075        }
1076    }
1077
1078    #[must_use]
1079    pub const fn effect_deadline(&self) -> Option<&U64> {
1080        self.effect_deadline.as_ref()
1081    }
1082
1083    #[must_use]
1084    pub const fn resource(&self) -> Option<&Resource> {
1085        self.resource.as_ref()
1086    }
1087
1088    #[must_use]
1089    pub const fn authority_binding(&self) -> Option<&(Identity, Identity, Digest, U64, U64, U64)> {
1090        self.binding.as_ref()
1091    }
1092
1093    #[must_use]
1094    pub const fn catalog_preparation(&self) -> Option<&(Locator, Digest, U64, IndexResourceSpec)> {
1095        self.catalog_preparation.as_ref()
1096    }
1097
1098    #[must_use]
1099    pub const fn index_active(&self) -> bool {
1100        self.index_active
1101    }
1102
1103    #[must_use]
1104    pub fn role_preparation(
1105        &self,
1106        target: DevelopmentReconcileRoleTarget,
1107    ) -> Option<&(Locator, Digest, U64, U64, U64)> {
1108        self.role_preparations.get(&target)
1109    }
1110
1111    #[must_use]
1112    pub fn role_applied(&self, target: DevelopmentReconcileRoleTarget) -> Option<&U64> {
1113        self.role_applied.get(&target)
1114    }
1115
1116    #[must_use]
1117    pub fn role_verified(&self, target: DevelopmentReconcileRoleTarget) -> Option<&(U64, U64)> {
1118        self.role_verified.get(&target)
1119    }
1120
1121    #[must_use]
1122    pub const fn reservation(&self) -> Option<&ServiceCredentialReservation> {
1123        self.reservation.as_ref()
1124    }
1125
1126    #[must_use]
1127    pub const fn credential_staged(&self) -> Option<&PublishedFileIdentity> {
1128        self.credential_staged.as_ref()
1129    }
1130
1131    #[must_use]
1132    pub const fn credential_activated(&self) -> bool {
1133        self.credential_activated
1134    }
1135
1136    #[must_use]
1137    pub const fn credential_published(&self) -> Option<&PublishedFileIdentity> {
1138        self.credential_published.as_ref()
1139    }
1140
1141    #[must_use]
1142    pub const fn profile_published(&self) -> Option<&PublishedFileIdentity> {
1143        self.profile_published.as_ref()
1144    }
1145
1146    #[must_use]
1147    pub const fn consumer_policy_revision(&self) -> Option<&U64> {
1148        self.consumer_verified.as_ref()
1149    }
1150
1151    #[must_use]
1152    pub const fn material_published(&self) -> Option<&PublishedFileIdentity> {
1153        self.material_published.as_ref()
1154    }
1155
1156    #[must_use]
1157    pub const fn terminal(&self) -> Option<DevelopmentReconcileTerminalOutcome> {
1158        self.terminal
1159    }
1160
1161    #[must_use]
1162    pub const fn completed(&self) -> bool {
1163        self.completed
1164    }
1165}
1166
1167fn valid_absolute_path(value: &str) -> bool {
1168    if value.is_empty() || value.len() > 4_096 || value.bytes().any(|byte| byte.is_ascii_control()) {
1169        return false;
1170    }
1171    let path = Path::new(value);
1172    path.is_absolute()
1173        && path.parent().is_some()
1174        && path
1175            .components()
1176            .all(|component| matches!(component, Component::RootDir | Component::Normal(_)))
1177        && path.components().collect::<PathBuf>().as_os_str() == path.as_os_str()
1178}
1179
1180fn valid_sha256(value: &[u8; 32]) -> bool {
1181    value.iter().any(|byte| *byte != 0)
1182}
1183
1184fn valid_instance_identity(value: &str) -> bool {
1185    value.len() == 64
1186        && value
1187            .bytes()
1188            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
1189        && value.bytes().any(|byte| byte != b'0')
1190}
1191
1192fn valid_git_object(value: &str) -> bool {
1193    matches!(value.len(), 40 | 64)
1194        && value
1195            .bytes()
1196            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
1197        && value.bytes().any(|byte| byte != b'0')
1198}
1199
1200fn valid_development_endpoints(native: &str, subscription: &str, internal: &str, bootstrap: &str) -> bool {
1201    let values = [native, subscription, internal, bootstrap];
1202    if values.iter().any(|value| {
1203        value.is_empty() || value.len() > MAX_ENDPOINT_BYTES || !value.bytes().all(|byte| byte.is_ascii_graphic())
1204    }) {
1205        return false;
1206    }
1207    let Some(authority) = native.strip_prefix("lodestone+tls://") else {
1208        return false;
1209    };
1210    let (advertised_host, port_text) = if let Some(port) = authority.strip_prefix("[::1]:") {
1211        ("[::1]", port)
1212    } else if let Some(port) = authority.strip_prefix("localhost:") {
1213        ("localhost", port)
1214    } else if let Some(port) = authority.strip_prefix("127.0.0.1:") {
1215        ("127.0.0.1", port)
1216    } else {
1217        return false;
1218    };
1219    let Some(base_port) = port_text
1220        .parse::<u16>()
1221        .ok()
1222        .filter(|port| port.to_string() == port_text && (1024..=65532).contains(port))
1223    else {
1224        return false;
1225    };
1226    let bind_host = match advertised_host {
1227        "[::1]" => "[::1]",
1228        _ => "127.0.0.1",
1229    };
1230    subscription == format!("https://{advertised_host}:{}", base_port + 1)
1231        && internal == format!("{bind_host}:{}", base_port + 2)
1232        && bootstrap == format!("{bind_host}:{}", base_port + 3)
1233}
1234
1235fn valid_identity(value: &Identity) -> bool {
1236    value.value.len() == 32
1237        && value
1238            .value
1239            .bytes()
1240            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
1241        && value.value.bytes().any(|byte| byte != b'0')
1242}
1243
1244fn valid_u64(value: &U64) -> bool {
1245    value
1246        .value
1247        .parse::<u64>()
1248        .ok()
1249        .is_some_and(|parsed| parsed.to_string() == value.value)
1250}
1251
1252fn canonical_u64(value: &U64) -> Option<u64> {
1253    value
1254        .value
1255        .parse::<u64>()
1256        .ok()
1257        .filter(|parsed| parsed.to_string() == value.value)
1258}
1259
1260fn ordered_successor(predecessor: &U64, successor: &U64) -> bool {
1261    canonical_u64(predecessor)
1262        .filter(|value| *value > 0)
1263        .and_then(|value| value.checked_add(1))
1264        .is_some_and(|expected| canonical_u64(successor) == Some(expected))
1265}
1266
1267fn ordered_u64(predecessor: &U64, successor: &U64) -> bool {
1268    canonical_u64(predecessor)
1269        .zip(canonical_u64(successor))
1270        .is_some_and(|(predecessor, successor)| predecessor < successor)
1271}
1272
1273fn ordered_or_equal_u64(predecessor: &U64, successor: &U64) -> bool {
1274    canonical_u64(predecessor)
1275        .zip(canonical_u64(successor))
1276        .is_some_and(|(predecessor, successor)| predecessor <= successor)
1277}
1278
1279fn checked_add_u64(left: &U64, right: &U64) -> Option<u64> {
1280    canonical_u64(left)?.checked_add(canonical_u64(right)?)
1281}
1282
1283fn valid_positive_bounded_duration(value: &U64) -> bool {
1284    canonical_u64(value).is_some_and(|value| (1..=86_400_000).contains(&value))
1285}
1286
1287fn valid_digest(value: &Digest) -> bool {
1288    value.value.len() == 64
1289        && value
1290            .value
1291            .bytes()
1292            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
1293        && value.value.bytes().any(|byte| byte != b'0')
1294}
1295
1296fn valid_locator(value: &Locator) -> bool {
1297    !value.authority.is_empty()
1298        && value.authority.len() <= MAX_NAME_BYTES
1299        && value.authority.bytes().all(|byte| byte.is_ascii_graphic())
1300        && !value.value.is_empty()
1301        && value.value.len() <= 4_096
1302        && value.value.bytes().all(|byte| byte.is_ascii_graphic())
1303}
1304
1305fn valid_resource(value: &Resource) -> bool {
1306    value.kind == "index" && valid_index_name(&value.name) && valid_identity(&value.incarnation)
1307}
1308
1309fn valid_index_name(value: &str) -> bool {
1310    !value.is_empty()
1311        && value.len() <= MAX_NAME_BYTES
1312        && value
1313            .bytes()
1314            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'-' | b'_'))
1315}
1316
1317fn valid_reservation(value: &ServiceCredentialReservation) -> bool {
1318    valid_locator(&value.locator)
1319        && value.locator.authority == "service_credential_reservation"
1320        && valid_digest(&value.reservation_action_id)
1321        && valid_digest(&value.activation_action_id)
1322        && valid_identity(&value.principal_id)
1323        && valid_identity(&value.api_key_id)
1324        && valid_u64(&value.generation)
1325        && valid_u64(&value.reservation_expires_at)
1326        && valid_u64(&value.credential_expires_at)
1327        && value.verifier_algorithm == "sha256_identity_bound_v1"
1328        && value.verifier_algorithm_version == 1
1329        && value.required_secret_bytes == 32
1330}
1331
1332#[cfg(test)]
1333#[allow(clippy::expect_used)] // Fixed nonzero byte fixtures are valid by construction.
1334mod tests {
1335    use super::*;
1336
1337    fn identity(value: u128) -> Identity {
1338        Identity {
1339            value: format!("{value:032x}"),
1340        }
1341    }
1342
1343    fn instance_identity(value: u8) -> String {
1344        format!("{value:02x}").repeat(32)
1345    }
1346
1347    fn digest(value: u8) -> Digest {
1348        Digest {
1349            value: format!("{value:02x}").repeat(32),
1350        }
1351    }
1352
1353    fn u64_value(value: u64) -> U64 {
1354        U64 {
1355            value: value.to_string(),
1356        }
1357    }
1358
1359    fn file(value: u64) -> PublishedFileIdentity {
1360        PublishedFileIdentity {
1361            device: 1,
1362            inode: value,
1363            size: 1,
1364            sha256: [u8::try_from(value).unwrap_or(1); 32],
1365        }
1366    }
1367
1368    fn reconcile_intent() -> DevelopmentReconcileIntent {
1369        DevelopmentReconcileIntent {
1370            format_version: DEVELOPMENT_RECONCILE_FORMAT_VERSION,
1371            instance_root: "/tmp/lodestone".to_owned(),
1372            instance_manifest_sha256: [1; 32],
1373            index_definition_path: "/tmp/lodestone/operator/development-index.json".to_owned(),
1374            index_definition_sha256: [2; 32],
1375            credential_output: "/tmp/output/credential.json".to_owned(),
1376            profile_output: "/tmp/output/profile.json".to_owned(),
1377            material_output: "/tmp/output/material.json".to_owned(),
1378            timeout_millis: u64_value(90),
1379            reservation_ttl_millis: u64_value(10),
1380            credential_ttl_millis: u64_value(20),
1381        }
1382    }
1383
1384    #[test]
1385    fn materialization_manifest_is_last_and_completion_requires_verification() -> Result<(), WorkflowJournalError> {
1386        let mut transitions = vec![
1387            DevelopmentMaterializeTransition::InstanceIdentityRecorded {
1388                instance_identity: instance_identity(1),
1389            },
1390            DevelopmentMaterializeTransition::ClockRecorded {
1391                unix_millis: u64_value(1),
1392            },
1393            DevelopmentMaterializeTransition::PrivatePlanStaged {
1394                plan: DevelopmentMaterializePrivatePlan::Pki,
1395                file: file(200),
1396            },
1397            DevelopmentMaterializeTransition::PrivatePlanPublished {
1398                plan: DevelopmentMaterializePrivatePlan::Pki,
1399                file: file(200),
1400            },
1401        ];
1402        for (position, artifact) in DevelopmentMaterializeArtifact::required_before_manifest()
1403            .into_iter()
1404            .enumerate()
1405        {
1406            let file = file(u64::try_from(position).unwrap_or(0) + 1);
1407            transitions.push(DevelopmentMaterializeTransition::ArtifactStaged {
1408                artifact,
1409                file: file.clone(),
1410            });
1411            transitions.push(DevelopmentMaterializeTransition::ArtifactPublished {
1412                artifact,
1413                file,
1414                public_sha256: None,
1415            });
1416        }
1417        let manifest = file(100);
1418        transitions.push(DevelopmentMaterializeTransition::ArtifactStaged {
1419            artifact: DevelopmentMaterializeArtifact::InstanceManifest,
1420            file: manifest.clone(),
1421        });
1422        transitions.push(DevelopmentMaterializeTransition::ArtifactPublished {
1423            artifact: DevelopmentMaterializeArtifact::InstanceManifest,
1424            file: manifest,
1425            public_sha256: Some([3; 32]),
1426        });
1427        assert!(DevelopmentMaterializeState::replay(&transitions).is_ok());
1428        transitions.push(DevelopmentMaterializeTransition::Verified {
1429            manifest_sha256: [3; 32],
1430        });
1431        transitions.push(DevelopmentMaterializeTransition::Completed);
1432        assert!(DevelopmentMaterializeState::replay(&transitions)?.completed());
1433
1434        let out_of_order = [
1435            DevelopmentMaterializeTransition::InstanceIdentityRecorded {
1436                instance_identity: instance_identity(1),
1437            },
1438            DevelopmentMaterializeTransition::ClockRecorded {
1439                unix_millis: u64_value(1),
1440            },
1441            DevelopmentMaterializeTransition::ArtifactPublished {
1442                artifact: DevelopmentMaterializeArtifact::InstanceManifest,
1443                file: file(1),
1444                public_sha256: Some([3; 32]),
1445            },
1446        ];
1447        assert_eq!(
1448            DevelopmentMaterializeState::replay(&out_of_order),
1449            Err(WorkflowJournalError::Corrupt)
1450        );
1451        Ok(())
1452    }
1453
1454    #[test]
1455    fn pki_successor_requires_ca_artifacts_only_for_ca_rotation() -> Result<(), WorkflowJournalError> {
1456        let intent = DevelopmentPkiSuccessorIntent {
1457            format_version: DEVELOPMENT_PKI_SUCCESSOR_FORMAT_VERSION,
1458            operation: DevelopmentPkiSuccessorOperation::RenewLeaves,
1459            instance_identity: instance_identity(1),
1460            parent_manifest_path: "/tmp/parent/instance.json".to_owned(),
1461            parent_manifest_sha256: [1; 32],
1462            successor_root: "/tmp/successor".to_owned(),
1463            parent_instance_generation: u64_value(1),
1464            successor_instance_generation: u64_value(2),
1465            parent_ca_generation: u64_value(1),
1466            successor_ca_generation: u64_value(1),
1467            parent_process_leaf_generation: u64_value(1),
1468            successor_process_leaf_generation: u64_value(2),
1469            parent_gateway_leaf_generation: u64_value(1),
1470            successor_gateway_leaf_generation: u64_value(2),
1471        };
1472        intent.validate()?;
1473        let illegal = [DevelopmentPkiSuccessorTransition::ArtifactStaged {
1474            artifact: DevelopmentPkiSuccessorArtifact::CaPrivateKey,
1475            file: file(1),
1476        }];
1477        assert_eq!(
1478            DevelopmentPkiSuccessorState::replay(&intent, &illegal),
1479            Err(WorkflowJournalError::Corrupt)
1480        );
1481
1482        let mut transitions = vec![
1483            DevelopmentPkiSuccessorTransition::ClockRecorded {
1484                unix_millis: u64_value(1),
1485            },
1486            DevelopmentPkiSuccessorTransition::PrivatePlanStaged {
1487                plan: DevelopmentPkiSuccessorPrivatePlan::Pki,
1488                file: file(90),
1489            },
1490            DevelopmentPkiSuccessorTransition::PrivatePlanPublished {
1491                plan: DevelopmentPkiSuccessorPrivatePlan::Pki,
1492                file: file(90),
1493            },
1494        ];
1495        for (position, artifact) in required_pki_artifacts(intent.operation).into_iter().enumerate() {
1496            let identity = file(u64::try_from(position).unwrap_or(0) + 1);
1497            transitions.push(DevelopmentPkiSuccessorTransition::ArtifactStaged {
1498                artifact,
1499                file: identity.clone(),
1500            });
1501            transitions.push(DevelopmentPkiSuccessorTransition::ArtifactPublished {
1502                artifact,
1503                file: identity,
1504                public_sha256: None,
1505            });
1506        }
1507        let manifest = file(100);
1508        transitions.push(DevelopmentPkiSuccessorTransition::ArtifactStaged {
1509            artifact: DevelopmentPkiSuccessorArtifact::InstanceManifest,
1510            file: manifest.clone(),
1511        });
1512        transitions.push(DevelopmentPkiSuccessorTransition::ArtifactPublished {
1513            artifact: DevelopmentPkiSuccessorArtifact::InstanceManifest,
1514            file: manifest,
1515            public_sha256: Some([9; 32]),
1516        });
1517        transitions.push(DevelopmentPkiSuccessorTransition::Verified {
1518            manifest_sha256: [9; 32],
1519        });
1520        transitions.push(DevelopmentPkiSuccessorTransition::Completed);
1521        assert!(DevelopmentPkiSuccessorState::replay(&intent, &transitions)?.completed());
1522        Ok(())
1523    }
1524
1525    #[test]
1526    fn reconciliation_enforces_administrator_scope_before_consumer_scope() -> Result<(), WorkflowJournalError> {
1527        let resource = Resource {
1528            kind: "index".to_owned(),
1529            name: "events".to_owned(),
1530            incarnation: identity(4),
1531        };
1532        let prefix = [
1533            DevelopmentReconcileTransition::EffectDeadlineRecorded {
1534                unix_millis: u64_value(100),
1535            },
1536            DevelopmentReconcileTransition::AuthorityObserved {
1537                cluster_id: identity(1),
1538                collection_id: identity(2),
1539                schema_sha256: digest(3),
1540                logical_time: u64_value(10),
1541                reservation_expires_at: u64_value(20),
1542                credential_expires_at: u64_value(30),
1543            },
1544            DevelopmentReconcileTransition::IndexActive {
1545                resource,
1546                index_revision: u64_value(1),
1547            },
1548            DevelopmentReconcileTransition::RoleVerified {
1549                target: DevelopmentReconcileRoleTarget::Consumer,
1550                role_version: u64_value(1),
1551                observed_policy_revision: u64_value(1),
1552            },
1553        ];
1554        assert_eq!(
1555            DevelopmentReconcileState::replay(&reconcile_intent(), &prefix),
1556            Err(WorkflowJournalError::Corrupt)
1557        );
1558        Ok(())
1559    }
1560
1561    #[test]
1562    fn reconciliation_terminal_outcome_is_final_but_unresolved_work_remains_resumable() {
1563        let deadline = DevelopmentReconcileTransition::EffectDeadlineRecorded {
1564            unix_millis: u64_value(100),
1565        };
1566        let terminal = DevelopmentReconcileTransition::Terminal {
1567            outcome: DevelopmentReconcileTerminalOutcome::EffectDeadlineExpired,
1568        };
1569        let state = DevelopmentReconcileState::replay(&reconcile_intent(), &[deadline.clone(), terminal]);
1570        assert_eq!(
1571            state.map(|value| value.terminal()),
1572            Ok(Some(DevelopmentReconcileTerminalOutcome::EffectDeadlineExpired))
1573        );
1574        assert_eq!(
1575            DevelopmentReconcileState::replay(
1576                &reconcile_intent(),
1577                &[
1578                    deadline,
1579                    DevelopmentReconcileTransition::Terminal {
1580                        outcome: DevelopmentReconcileTerminalOutcome::Conflict,
1581                    },
1582                    DevelopmentReconcileTransition::Completed,
1583                ],
1584            ),
1585            Err(WorkflowJournalError::Corrupt)
1586        );
1587
1588        // An unresolved remote locator is already retained by its preparation
1589        // transition. No terminal record is appended, so a later invocation can
1590        // spend a fresh observation budget resolving the same locator.
1591        let resumable = [
1592            DevelopmentReconcileTransition::EffectDeadlineRecorded {
1593                unix_millis: u64_value(100),
1594            },
1595            DevelopmentReconcileTransition::AuthorityObserved {
1596                cluster_id: identity(1),
1597                collection_id: identity(2),
1598                schema_sha256: digest(3),
1599                logical_time: u64_value(10),
1600                reservation_expires_at: u64_value(20),
1601                credential_expires_at: u64_value(30),
1602            },
1603            DevelopmentReconcileTransition::CatalogPrepared {
1604                locator: Locator {
1605                    authority: "catalog".to_owned(),
1606                    value: "create-events".to_owned(),
1607                },
1608                semantic_digest: digest(4),
1609                expires_at: u64_value(40),
1610                resource_spec: IndexResourceSpec {
1611                    format_version: u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION),
1612                    revision: u64_value(1),
1613                    initial_shards: 1,
1614                    target_shard_bytes: u64_value(1),
1615                    durability: crate::ResourceDurabilityClass::RegionalObject,
1616                    storage_class: crate::ResourceStorageClass::Standard,
1617                    ram_admission_weight: 1,
1618                    ram_hard_max_bytes: u64_value(1),
1619                    nvme_admission_weight: 1,
1620                    nvme_hard_max_bytes: u64_value(1),
1621                    refresh_objective_millis: 1,
1622                    memtable_hard_max_bytes: u64_value(1),
1623                    seal_parallelism: 1,
1624                    merge_parallelism: 1,
1625                    vector: lodestone_wire::VectorResourcePolicy {
1626                        quality: lodestone_wire::VectorQualityObjective::Balanced,
1627                        indexing: lodestone_wire::VectorIndexingObjective::Balanced,
1628                        residency: None,
1629                        calibration: None,
1630                    },
1631                    lifecycle_binding: None,
1632                    encryption_required: false,
1633                    resolver_capability: None,
1634                },
1635            },
1636        ];
1637        let state = DevelopmentReconcileState::replay(&reconcile_intent(), &resumable);
1638        assert!(state.is_ok_and(|value| value.terminal().is_none() && value.catalog_preparation().is_some()));
1639    }
1640}