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