1use 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)] mod 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 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}