use std::collections::{BTreeMap, BTreeSet};
use std::path::{Component, Path, PathBuf};
use lodestone_wire::{
Digest, Identity, IndexResourceSpec, LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION, Locator, Resource,
ServiceCredentialReservation, U64,
};
use serde::{Deserialize, Serialize};
use crate::{
LocalOperatorWorkflowKind, PublishedFileIdentity, WorkflowJournalError, WorkflowJournalIntent,
WorkflowJournalTransition,
};
pub const DEVELOPMENT_MATERIALIZE_FORMAT_VERSION: u16 = 1;
pub const DEVELOPMENT_PKI_SUCCESSOR_FORMAT_VERSION: u16 = 1;
pub const DEVELOPMENT_RECONCILE_FORMAT_VERSION: u16 = 1;
const MAX_ENDPOINT_BYTES: usize = 2_048;
const MAX_NAME_BYTES: usize = 255;
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DevelopmentMaterializeIntent {
pub format_version: u16,
pub output_root: String,
pub supply_root: String,
pub supply_package_sha256: [u8; 32],
pub supply_source_revision: String,
pub native_api_sha256: [u8; 32],
pub native_endpoint: String,
pub subscription_endpoint: String,
pub internal_endpoint: String,
pub bootstrap_endpoint: String,
}
impl crate::workflow_journal::sealed::Sealed for DevelopmentMaterializeIntent {}
impl WorkflowJournalIntent for DevelopmentMaterializeIntent {
type Transition = DevelopmentMaterializeTransition;
const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::DevelopmentMaterialize;
fn validate(&self) -> Result<(), WorkflowJournalError> {
if self.format_version != DEVELOPMENT_MATERIALIZE_FORMAT_VERSION
|| !valid_absolute_path(&self.output_root)
|| !valid_absolute_path(&self.supply_root)
|| self.output_root == self.supply_root
|| !valid_sha256(&self.supply_package_sha256)
|| !valid_git_object(&self.supply_source_revision)
|| !valid_sha256(&self.native_api_sha256)
|| !valid_development_endpoints(
&self.native_endpoint,
&self.subscription_endpoint,
&self.internal_endpoint,
&self.bootstrap_endpoint,
)
{
return Err(WorkflowJournalError::InvalidIntent);
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentMaterializeArtifact {
CaCertificate,
CaPrivateKey,
ProcessCertificate,
ProcessPrivateKey,
GatewayCertificate,
GatewayPrivateKey,
BootstrapCredential,
BootstrapIntent,
ServerConfiguration,
DevelopmentIndexDefinition,
InstanceManifest,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentMaterializePrivatePlan {
Pki,
}
impl DevelopmentMaterializeArtifact {
fn is_manifest(self) -> bool {
self == Self::InstanceManifest
}
fn required_before_manifest() -> BTreeSet<Self> {
[
Self::CaCertificate,
Self::CaPrivateKey,
Self::ProcessCertificate,
Self::ProcessPrivateKey,
Self::GatewayCertificate,
Self::GatewayPrivateKey,
Self::BootstrapCredential,
Self::BootstrapIntent,
Self::ServerConfiguration,
Self::DevelopmentIndexDefinition,
]
.into_iter()
.collect()
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum DevelopmentMaterializeTransition {
InstanceIdentityRecorded {
instance_identity: String,
},
ClockRecorded {
unix_millis: U64,
},
PrivatePlanStaged {
plan: DevelopmentMaterializePrivatePlan,
file: PublishedFileIdentity,
},
PrivatePlanPublished {
plan: DevelopmentMaterializePrivatePlan,
file: PublishedFileIdentity,
},
ArtifactStaged {
artifact: DevelopmentMaterializeArtifact,
file: PublishedFileIdentity,
},
ArtifactPublished {
artifact: DevelopmentMaterializeArtifact,
file: PublishedFileIdentity,
#[serde(default, skip_serializing_if = "Option::is_none")]
public_sha256: Option<[u8; 32]>,
},
Verified {
manifest_sha256: [u8; 32],
},
Completed,
}
impl crate::workflow_journal::sealed::Sealed for DevelopmentMaterializeTransition {}
impl WorkflowJournalTransition for DevelopmentMaterializeTransition {
fn validate(&self) -> Result<(), WorkflowJournalError> {
let valid = match self {
Self::InstanceIdentityRecorded { instance_identity } => valid_instance_identity(instance_identity),
Self::ClockRecorded { unix_millis } => valid_u64(unix_millis),
Self::PrivatePlanStaged { file, .. }
| Self::PrivatePlanPublished { file, .. }
| Self::ArtifactStaged { file, .. } => file.is_valid(),
Self::ArtifactPublished {
artifact,
file,
public_sha256,
} => {
file.is_valid()
&& public_sha256.as_ref().is_none_or(valid_sha256)
&& (!artifact.is_manifest() || public_sha256.as_ref().is_some_and(valid_sha256))
}
Self::Verified { manifest_sha256 } => valid_sha256(manifest_sha256),
Self::Completed => true,
};
if valid {
Ok(())
} else {
Err(WorkflowJournalError::InvalidTransition)
}
}
}
#[derive(Clone, Debug, Default, PartialEq)]
pub struct DevelopmentMaterializeState {
instance_identity: Option<String>,
unix_millis: Option<U64>,
private_plans: BTreeMap<DevelopmentMaterializePrivatePlan, PublishedFileIdentity>,
published_private_plans: BTreeMap<DevelopmentMaterializePrivatePlan, PublishedFileIdentity>,
staged: BTreeMap<DevelopmentMaterializeArtifact, PublishedFileIdentity>,
published: BTreeMap<DevelopmentMaterializeArtifact, PublishedFileIdentity>,
published_manifest_sha256: Option<[u8; 32]>,
manifest_sha256: Option<[u8; 32]>,
completed: bool,
}
impl DevelopmentMaterializeState {
pub fn replay(transitions: &[DevelopmentMaterializeTransition]) -> Result<Self, WorkflowJournalError> {
let mut state = Self::default();
for transition in transitions {
transition.validate()?;
state.apply(transition)?;
}
Ok(state)
}
fn apply(&mut self, transition: &DevelopmentMaterializeTransition) -> Result<(), WorkflowJournalError> {
if self.completed {
return Err(WorkflowJournalError::Corrupt);
}
match transition {
DevelopmentMaterializeTransition::InstanceIdentityRecorded { instance_identity }
if self.instance_identity.is_none() && self.unix_millis.is_none() =>
{
self.instance_identity = Some(instance_identity.clone());
}
DevelopmentMaterializeTransition::ClockRecorded { unix_millis }
if self.instance_identity.is_some() && self.unix_millis.is_none() =>
{
self.unix_millis = Some(unix_millis.clone());
}
DevelopmentMaterializeTransition::PrivatePlanStaged { plan, file }
if self.unix_millis.is_some()
&& self.staged.is_empty()
&& self.published.is_empty()
&& !self.private_plans.contains_key(plan) =>
{
self.private_plans.insert(*plan, file.clone());
}
DevelopmentMaterializeTransition::PrivatePlanPublished { plan, file }
if !self.published_private_plans.contains_key(plan) && self.private_plans.get(plan) == Some(file) =>
{
self.published_private_plans.insert(*plan, file.clone());
}
DevelopmentMaterializeTransition::ArtifactStaged { artifact, file }
if self
.published_private_plans
.contains_key(&DevelopmentMaterializePrivatePlan::Pki)
&& !self.staged.contains_key(artifact)
&& !self.published.contains_key(artifact)
&& (!artifact.is_manifest()
|| self.published.keys().copied().collect::<BTreeSet<_>>()
== DevelopmentMaterializeArtifact::required_before_manifest()) =>
{
self.staged.insert(*artifact, file.clone());
}
DevelopmentMaterializeTransition::ArtifactPublished {
artifact,
file,
public_sha256,
} if !self.published.contains_key(artifact) && self.staged.get(artifact) == Some(file) => {
self.published.insert(*artifact, file.clone());
if artifact.is_manifest() {
self.published_manifest_sha256 = *public_sha256;
}
}
DevelopmentMaterializeTransition::Verified { manifest_sha256 }
if self
.published
.contains_key(&DevelopmentMaterializeArtifact::InstanceManifest)
&& self.published_manifest_sha256.as_ref() == Some(manifest_sha256)
&& self.manifest_sha256.is_none() =>
{
self.manifest_sha256 = Some(*manifest_sha256);
}
DevelopmentMaterializeTransition::Completed if self.manifest_sha256.is_some() => {
self.completed = true;
}
_ => return Err(WorkflowJournalError::Corrupt),
}
Ok(())
}
#[must_use]
pub fn instance_identity(&self) -> Option<&str> {
self.instance_identity.as_deref()
}
#[must_use]
pub const fn clock(&self) -> Option<&U64> {
self.unix_millis.as_ref()
}
#[must_use]
pub fn staged(&self, artifact: DevelopmentMaterializeArtifact) -> Option<&PublishedFileIdentity> {
self.staged.get(&artifact)
}
#[must_use]
pub fn private_plan(&self, plan: DevelopmentMaterializePrivatePlan) -> Option<&PublishedFileIdentity> {
self.private_plans.get(&plan)
}
#[must_use]
pub fn published_private_plan(&self, plan: DevelopmentMaterializePrivatePlan) -> Option<&PublishedFileIdentity> {
self.published_private_plans.get(&plan)
}
#[must_use]
pub fn published(&self, artifact: DevelopmentMaterializeArtifact) -> Option<&PublishedFileIdentity> {
self.published.get(&artifact)
}
#[must_use]
pub const fn completed(&self) -> bool {
self.completed
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentPkiSuccessorOperation {
RenewLeaves,
RotateCa,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DevelopmentPkiSuccessorIntent {
pub format_version: u16,
pub operation: DevelopmentPkiSuccessorOperation,
pub instance_identity: String,
pub parent_manifest_path: String,
pub parent_manifest_sha256: [u8; 32],
pub successor_root: String,
pub parent_instance_generation: U64,
pub successor_instance_generation: U64,
pub parent_ca_generation: U64,
pub successor_ca_generation: U64,
pub parent_process_leaf_generation: U64,
pub successor_process_leaf_generation: U64,
pub parent_gateway_leaf_generation: U64,
pub successor_gateway_leaf_generation: U64,
}
impl crate::workflow_journal::sealed::Sealed for DevelopmentPkiSuccessorIntent {}
impl WorkflowJournalIntent for DevelopmentPkiSuccessorIntent {
type Transition = DevelopmentPkiSuccessorTransition;
const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::DevelopmentPkiSuccessor;
fn validate(&self) -> Result<(), WorkflowJournalError> {
let instance_generations =
ordered_successor(&self.parent_instance_generation, &self.successor_instance_generation);
let process_generations = ordered_successor(
&self.parent_process_leaf_generation,
&self.successor_process_leaf_generation,
);
let gateway_generations = ordered_successor(
&self.parent_gateway_leaf_generation,
&self.successor_gateway_leaf_generation,
);
let ca_generations = match self.operation {
DevelopmentPkiSuccessorOperation::RenewLeaves => {
canonical_u64(&self.parent_ca_generation).filter(|value| *value > 0)
== canonical_u64(&self.successor_ca_generation).filter(|value| *value > 0)
}
DevelopmentPkiSuccessorOperation::RotateCa => {
ordered_successor(&self.parent_ca_generation, &self.successor_ca_generation)
}
};
let parent_manifest = Path::new(&self.parent_manifest_path);
let parent_root = parent_manifest.parent();
if self.format_version != DEVELOPMENT_PKI_SUCCESSOR_FORMAT_VERSION
|| !valid_instance_identity(&self.instance_identity)
|| !valid_absolute_path(&self.parent_manifest_path)
|| !valid_sha256(&self.parent_manifest_sha256)
|| !valid_absolute_path(&self.successor_root)
|| parent_manifest.file_name().and_then(|name| name.to_str()) != Some("instance.json")
|| parent_root.is_none_or(|root| paths_overlap(root, Path::new(&self.successor_root)))
|| !instance_generations
|| !process_generations
|| !gateway_generations
|| !ca_generations
{
return Err(WorkflowJournalError::InvalidIntent);
}
Ok(())
}
}
fn paths_overlap(left: &Path, right: &Path) -> bool {
left.starts_with(right) || right.starts_with(left)
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentPkiSuccessorArtifact {
CaCertificate,
CaPrivateKey,
ProcessCertificate,
ProcessPrivateKey,
GatewayCertificate,
GatewayPrivateKey,
ServerConfiguration,
InstanceManifest,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentPkiSuccessorPrivatePlan {
Pki,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum DevelopmentPkiSuccessorTransition {
ClockRecorded {
unix_millis: U64,
},
PrivatePlanStaged {
plan: DevelopmentPkiSuccessorPrivatePlan,
file: PublishedFileIdentity,
},
PrivatePlanPublished {
plan: DevelopmentPkiSuccessorPrivatePlan,
file: PublishedFileIdentity,
},
ArtifactStaged {
artifact: DevelopmentPkiSuccessorArtifact,
file: PublishedFileIdentity,
},
ArtifactPublished {
artifact: DevelopmentPkiSuccessorArtifact,
file: PublishedFileIdentity,
#[serde(default, skip_serializing_if = "Option::is_none")]
public_sha256: Option<[u8; 32]>,
},
Verified {
manifest_sha256: [u8; 32],
},
Completed,
}
impl crate::workflow_journal::sealed::Sealed for DevelopmentPkiSuccessorTransition {}
impl WorkflowJournalTransition for DevelopmentPkiSuccessorTransition {
fn validate(&self) -> Result<(), WorkflowJournalError> {
let valid = match self {
Self::ClockRecorded { unix_millis } => valid_u64(unix_millis),
Self::PrivatePlanStaged { file, .. }
| Self::PrivatePlanPublished { file, .. }
| Self::ArtifactStaged { file, .. } => file.is_valid(),
Self::ArtifactPublished {
artifact,
file,
public_sha256,
} => {
file.is_valid()
&& public_sha256.as_ref().is_none_or(valid_sha256)
&& (*artifact != DevelopmentPkiSuccessorArtifact::InstanceManifest
|| public_sha256.as_ref().is_some_and(valid_sha256))
}
Self::Verified { manifest_sha256 } => valid_sha256(manifest_sha256),
Self::Completed => true,
};
if valid {
Ok(())
} else {
Err(WorkflowJournalError::InvalidTransition)
}
}
}
#[derive(Clone, Debug, Default, PartialEq)]
pub struct DevelopmentPkiSuccessorState {
clock: Option<U64>,
private_plans: BTreeMap<DevelopmentPkiSuccessorPrivatePlan, PublishedFileIdentity>,
published_private_plans: BTreeMap<DevelopmentPkiSuccessorPrivatePlan, PublishedFileIdentity>,
staged: BTreeMap<DevelopmentPkiSuccessorArtifact, PublishedFileIdentity>,
published: BTreeMap<DevelopmentPkiSuccessorArtifact, PublishedFileIdentity>,
published_manifest_sha256: Option<[u8; 32]>,
manifest_sha256: Option<[u8; 32]>,
completed: bool,
}
impl DevelopmentPkiSuccessorState {
pub fn replay(
intent: &DevelopmentPkiSuccessorIntent,
transitions: &[DevelopmentPkiSuccessorTransition],
) -> Result<Self, WorkflowJournalError> {
intent.validate()?;
let mut state = Self::default();
for transition in transitions {
transition.validate()?;
state.apply(intent.operation, transition)?;
}
Ok(state)
}
fn apply(
&mut self,
operation: DevelopmentPkiSuccessorOperation,
transition: &DevelopmentPkiSuccessorTransition,
) -> Result<(), WorkflowJournalError> {
if self.completed {
return Err(WorkflowJournalError::Corrupt);
}
match transition {
DevelopmentPkiSuccessorTransition::ClockRecorded { unix_millis } if self.clock.is_none() => {
self.clock = Some(unix_millis.clone());
}
DevelopmentPkiSuccessorTransition::PrivatePlanStaged { plan, file }
if self.clock.is_some()
&& self.staged.is_empty()
&& self.published.is_empty()
&& !self.private_plans.contains_key(plan) =>
{
self.private_plans.insert(*plan, file.clone());
}
DevelopmentPkiSuccessorTransition::PrivatePlanPublished { plan, file }
if !self.published_private_plans.contains_key(plan) && self.private_plans.get(plan) == Some(file) =>
{
self.published_private_plans.insert(*plan, file.clone());
}
DevelopmentPkiSuccessorTransition::ArtifactStaged { artifact, file }
if self.clock.is_some()
&& self
.published_private_plans
.contains_key(&DevelopmentPkiSuccessorPrivatePlan::Pki)
&& !self.staged.contains_key(artifact)
&& !self.published.contains_key(artifact)
&& (artifact_admitted_for_pki_operation(operation, *artifact)
|| (*artifact == DevelopmentPkiSuccessorArtifact::InstanceManifest
&& self.published.keys().copied().collect::<BTreeSet<_>>()
== required_pki_artifacts(operation))) =>
{
self.staged.insert(*artifact, file.clone());
}
DevelopmentPkiSuccessorTransition::ArtifactPublished {
artifact,
file,
public_sha256,
} if !self.published.contains_key(artifact) && self.staged.get(artifact) == Some(file) => {
self.published.insert(*artifact, file.clone());
if *artifact == DevelopmentPkiSuccessorArtifact::InstanceManifest {
self.published_manifest_sha256 = *public_sha256;
}
}
DevelopmentPkiSuccessorTransition::Verified { manifest_sha256 }
if self
.published
.contains_key(&DevelopmentPkiSuccessorArtifact::InstanceManifest)
&& self.published_manifest_sha256.as_ref() == Some(manifest_sha256)
&& self.manifest_sha256.is_none() =>
{
self.manifest_sha256 = Some(*manifest_sha256);
}
DevelopmentPkiSuccessorTransition::Completed if self.manifest_sha256.is_some() => {
self.completed = true;
}
_ => return Err(WorkflowJournalError::Corrupt),
}
Ok(())
}
#[must_use]
pub const fn clock(&self) -> Option<&U64> {
self.clock.as_ref()
}
#[must_use]
pub fn private_plan(&self, plan: DevelopmentPkiSuccessorPrivatePlan) -> Option<&PublishedFileIdentity> {
self.private_plans.get(&plan)
}
#[must_use]
pub fn published_private_plan(&self, plan: DevelopmentPkiSuccessorPrivatePlan) -> Option<&PublishedFileIdentity> {
self.published_private_plans.get(&plan)
}
#[must_use]
pub fn staged(&self, artifact: DevelopmentPkiSuccessorArtifact) -> Option<&PublishedFileIdentity> {
self.staged.get(&artifact)
}
#[must_use]
pub fn published(&self, artifact: DevelopmentPkiSuccessorArtifact) -> Option<&PublishedFileIdentity> {
self.published.get(&artifact)
}
#[must_use]
pub const fn completed(&self) -> bool {
self.completed
}
}
fn artifact_admitted_for_pki_operation(
operation: DevelopmentPkiSuccessorOperation,
artifact: DevelopmentPkiSuccessorArtifact,
) -> bool {
artifact != DevelopmentPkiSuccessorArtifact::InstanceManifest
&& (operation == DevelopmentPkiSuccessorOperation::RotateCa
|| !matches!(
artifact,
DevelopmentPkiSuccessorArtifact::CaCertificate | DevelopmentPkiSuccessorArtifact::CaPrivateKey
))
}
fn required_pki_artifacts(operation: DevelopmentPkiSuccessorOperation) -> BTreeSet<DevelopmentPkiSuccessorArtifact> {
let mut required = [
DevelopmentPkiSuccessorArtifact::ProcessCertificate,
DevelopmentPkiSuccessorArtifact::ProcessPrivateKey,
DevelopmentPkiSuccessorArtifact::GatewayCertificate,
DevelopmentPkiSuccessorArtifact::GatewayPrivateKey,
DevelopmentPkiSuccessorArtifact::ServerConfiguration,
]
.into_iter()
.collect::<BTreeSet<_>>();
if operation == DevelopmentPkiSuccessorOperation::RotateCa {
required.insert(DevelopmentPkiSuccessorArtifact::CaCertificate);
required.insert(DevelopmentPkiSuccessorArtifact::CaPrivateKey);
}
required
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DevelopmentReconcileIntent {
pub format_version: u16,
pub instance_root: String,
pub instance_manifest_sha256: [u8; 32],
pub index_definition_path: String,
pub index_definition_sha256: [u8; 32],
pub credential_output: String,
pub profile_output: String,
pub material_output: String,
pub timeout_millis: U64,
pub reservation_ttl_millis: U64,
pub credential_ttl_millis: U64,
}
impl crate::workflow_journal::sealed::Sealed for DevelopmentReconcileIntent {}
impl WorkflowJournalIntent for DevelopmentReconcileIntent {
type Transition = DevelopmentReconcileTransition;
const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::DevelopmentReconcile;
fn validate(&self) -> Result<(), WorkflowJournalError> {
let paths = [
&self.instance_root,
&self.index_definition_path,
&self.credential_output,
&self.profile_output,
&self.material_output,
];
let distinct_outputs = [&self.credential_output, &self.profile_output, &self.material_output]
.into_iter()
.collect::<BTreeSet<_>>()
.len()
== 3;
if self.format_version != DEVELOPMENT_RECONCILE_FORMAT_VERSION
|| paths.into_iter().any(|path| !valid_absolute_path(path))
|| !distinct_outputs
|| !valid_sha256(&self.instance_manifest_sha256)
|| !valid_sha256(&self.index_definition_sha256)
|| !valid_positive_bounded_duration(&self.timeout_millis)
|| !valid_positive_bounded_duration(&self.reservation_ttl_millis)
|| !valid_positive_bounded_duration(&self.credential_ttl_millis)
|| !ordered_u64(&self.reservation_ttl_millis, &self.credential_ttl_millis)
{
return Err(WorkflowJournalError::InvalidIntent);
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentReconcileRoleTarget {
Administrator,
Consumer,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DevelopmentReconcileTerminalOutcome {
EffectDeadlineExpired,
Conflict,
ReservationExpiredNoEffect,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum DevelopmentReconcileTransition {
EffectDeadlineRecorded {
unix_millis: U64,
},
AuthorityObserved {
cluster_id: Identity,
collection_id: Identity,
schema_sha256: Digest,
logical_time: U64,
reservation_expires_at: U64,
credential_expires_at: U64,
},
CatalogPrepared {
locator: Locator,
semantic_digest: Digest,
expires_at: U64,
resource_spec: IndexResourceSpec,
},
CatalogApplied {
locator: Locator,
resource: Resource,
},
IndexActive {
resource: Resource,
index_revision: U64,
},
RolePrepared {
target: DevelopmentReconcileRoleTarget,
locator: Locator,
semantic_digest: Digest,
expires_at: U64,
predecessor_role_version: U64,
successor_role_version: U64,
},
RoleApplied {
target: DevelopmentReconcileRoleTarget,
locator: Locator,
target_policy_revision: U64,
},
RoleVerified {
target: DevelopmentReconcileRoleTarget,
role_version: U64,
observed_policy_revision: U64,
},
CredentialReserved {
reservation: ServiceCredentialReservation,
},
CredentialStaged {
file: PublishedFileIdentity,
},
CredentialActivated {
principal_id: Identity,
api_key_id: Identity,
generation: U64,
},
CredentialPublished {
file: PublishedFileIdentity,
},
ProfilePublished {
file: PublishedFileIdentity,
profile_sha256: [u8; 32],
},
ConsumerVerified {
observed_policy_revision: U64,
},
MaterialPublished {
file: PublishedFileIdentity,
material_sha256: [u8; 32],
},
Terminal {
outcome: DevelopmentReconcileTerminalOutcome,
},
Completed,
}
impl crate::workflow_journal::sealed::Sealed for DevelopmentReconcileTransition {}
impl WorkflowJournalTransition for DevelopmentReconcileTransition {
fn validate(&self) -> Result<(), WorkflowJournalError> {
let valid = match self {
Self::EffectDeadlineRecorded { unix_millis } => valid_u64(unix_millis),
Self::AuthorityObserved {
cluster_id,
collection_id,
schema_sha256,
logical_time,
reservation_expires_at,
credential_expires_at,
} => {
valid_identity(cluster_id)
&& valid_identity(collection_id)
&& valid_digest(schema_sha256)
&& valid_u64(logical_time)
&& ordered_u64(logical_time, reservation_expires_at)
&& ordered_u64(reservation_expires_at, credential_expires_at)
}
Self::CatalogPrepared {
locator,
semantic_digest,
expires_at,
resource_spec,
} => {
valid_locator(locator)
&& valid_digest(semantic_digest)
&& valid_u64(expires_at)
&& resource_spec.format_version == u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION)
}
Self::CatalogApplied { locator, resource } => valid_locator(locator) && valid_resource(resource),
Self::IndexActive {
resource,
index_revision,
} => valid_resource(resource) && valid_u64(index_revision),
Self::RolePrepared {
locator,
semantic_digest,
expires_at,
predecessor_role_version,
successor_role_version,
..
} => {
valid_locator(locator)
&& valid_digest(semantic_digest)
&& valid_u64(expires_at)
&& ordered_successor(predecessor_role_version, successor_role_version)
}
Self::RoleApplied {
locator,
target_policy_revision,
..
} => valid_locator(locator) && valid_u64(target_policy_revision),
Self::RoleVerified {
role_version,
observed_policy_revision,
..
} => valid_u64(role_version) && valid_u64(observed_policy_revision),
Self::CredentialReserved { reservation } => valid_reservation(reservation),
Self::CredentialStaged { file } | Self::CredentialPublished { file } => file.is_valid(),
Self::CredentialActivated {
principal_id,
api_key_id,
generation,
} => valid_identity(principal_id) && valid_identity(api_key_id) && valid_u64(generation),
Self::ProfilePublished { file, profile_sha256 } => file.is_valid() && valid_sha256(profile_sha256),
Self::ConsumerVerified {
observed_policy_revision,
} => valid_u64(observed_policy_revision),
Self::MaterialPublished { file, material_sha256 } => file.is_valid() && valid_sha256(material_sha256),
Self::Terminal { .. } => true,
Self::Completed => true,
};
if valid {
Ok(())
} else {
Err(WorkflowJournalError::InvalidTransition)
}
}
}
#[derive(Clone, Debug, Default, PartialEq)]
pub struct DevelopmentReconcileState {
effect_deadline: Option<U64>,
binding: Option<(Identity, Identity, Digest, U64, U64, U64)>,
catalog_preparation: Option<(Locator, Digest, U64, IndexResourceSpec)>,
resource: Option<Resource>,
index_active: bool,
role_preparations: BTreeMap<DevelopmentReconcileRoleTarget, (Locator, Digest, U64, U64, U64)>,
role_applied: BTreeMap<DevelopmentReconcileRoleTarget, U64>,
role_verified: BTreeMap<DevelopmentReconcileRoleTarget, (U64, U64)>,
reservation: Option<ServiceCredentialReservation>,
credential_staged: Option<PublishedFileIdentity>,
credential_activated: bool,
credential_published: Option<PublishedFileIdentity>,
profile_published: Option<PublishedFileIdentity>,
consumer_verified: Option<U64>,
material_published: Option<PublishedFileIdentity>,
terminal: Option<DevelopmentReconcileTerminalOutcome>,
completed: bool,
}
impl DevelopmentReconcileState {
pub fn replay(
intent: &DevelopmentReconcileIntent,
transitions: &[DevelopmentReconcileTransition],
) -> Result<Self, WorkflowJournalError> {
intent.validate()?;
let mut state = Self::default();
for transition in transitions {
transition.validate()?;
state.apply(intent, transition)?;
}
Ok(state)
}
fn apply(
&mut self,
intent: &DevelopmentReconcileIntent,
transition: &DevelopmentReconcileTransition,
) -> Result<(), WorkflowJournalError> {
if self.completed || self.terminal.is_some() {
return Err(WorkflowJournalError::Corrupt);
}
match transition {
DevelopmentReconcileTransition::EffectDeadlineRecorded { unix_millis }
if self.effect_deadline.is_none() =>
{
self.effect_deadline = Some(unix_millis.clone());
}
DevelopmentReconcileTransition::AuthorityObserved {
cluster_id,
collection_id,
schema_sha256,
logical_time,
reservation_expires_at,
credential_expires_at,
} if self.effect_deadline.is_some()
&& self.binding.is_none()
&& checked_add_u64(logical_time, &intent.reservation_ttl_millis)
== canonical_u64(reservation_expires_at)
&& checked_add_u64(logical_time, &intent.credential_ttl_millis)
== canonical_u64(credential_expires_at) =>
{
self.binding = Some((
cluster_id.clone(),
collection_id.clone(),
schema_sha256.clone(),
logical_time.clone(),
reservation_expires_at.clone(),
credential_expires_at.clone(),
));
}
DevelopmentReconcileTransition::CatalogPrepared {
locator,
semantic_digest,
expires_at,
resource_spec,
} if self.binding.is_some() && self.resource.is_none() => {
self.catalog_preparation = Some((
locator.clone(),
semantic_digest.clone(),
expires_at.clone(),
resource_spec.clone(),
));
}
DevelopmentReconcileTransition::CatalogApplied { locator, resource }
if self
.catalog_preparation
.as_ref()
.is_some_and(|(expected, _, _, _)| expected == locator)
&& self.resource.is_none() =>
{
self.resource = Some(resource.clone());
}
DevelopmentReconcileTransition::IndexActive { resource, .. }
if self.binding.is_some()
&& self.resource.as_ref().is_none_or(|current| current == resource)
&& !self.index_active =>
{
self.resource = Some(resource.clone());
self.index_active = true;
}
DevelopmentReconcileTransition::RolePrepared {
target,
locator,
semantic_digest,
expires_at,
predecessor_role_version,
successor_role_version,
} if self.role_target_is_next(*target) && !self.role_applied.contains_key(target) => {
self.role_preparations.insert(
*target,
(
locator.clone(),
semantic_digest.clone(),
expires_at.clone(),
predecessor_role_version.clone(),
successor_role_version.clone(),
),
);
}
DevelopmentReconcileTransition::RoleApplied {
target,
locator,
target_policy_revision,
} if self
.role_preparations
.get(target)
.is_some_and(|(expected, _, _, _, _)| expected == locator)
&& !self.role_applied.contains_key(target) =>
{
self.role_applied.insert(*target, target_policy_revision.clone());
}
DevelopmentReconcileTransition::RoleVerified {
target,
role_version,
observed_policy_revision,
} if self.role_target_is_next(*target)
&& self.role_verification_matches(*target, role_version, observed_policy_revision) =>
{
self.role_verified
.insert(*target, (role_version.clone(), observed_policy_revision.clone()));
}
DevelopmentReconcileTransition::CredentialReserved { reservation }
if self
.role_verified
.contains_key(&DevelopmentReconcileRoleTarget::Consumer)
&& self.reservation.is_none() =>
{
let Some((_, _, _, _, reservation_expires_at, credential_expires_at)) = self.binding.as_ref() else {
return Err(WorkflowJournalError::Corrupt);
};
if &reservation.reservation_expires_at != reservation_expires_at
|| &reservation.credential_expires_at != credential_expires_at
{
return Err(WorkflowJournalError::Corrupt);
}
self.reservation = Some(reservation.clone());
}
DevelopmentReconcileTransition::CredentialStaged { file }
if self.reservation.is_some() && self.credential_staged.is_none() =>
{
self.credential_staged = Some(file.clone());
}
DevelopmentReconcileTransition::CredentialActivated {
principal_id,
api_key_id,
generation,
} if self.reservation.as_ref().is_some_and(|reservation| {
&reservation.principal_id == principal_id
&& &reservation.api_key_id == api_key_id
&& &reservation.generation == generation
}) && self.credential_staged.is_some()
&& !self.credential_activated =>
{
self.credential_activated = true;
}
DevelopmentReconcileTransition::CredentialPublished { file }
if self.credential_activated
&& self.credential_staged.as_ref() == Some(file)
&& self.credential_published.is_none() =>
{
self.credential_published = Some(file.clone());
}
DevelopmentReconcileTransition::ProfilePublished { file, .. }
if self.credential_published.is_some() && self.profile_published.is_none() =>
{
self.profile_published = Some(file.clone());
}
DevelopmentReconcileTransition::ConsumerVerified {
observed_policy_revision,
} if self.profile_published.is_some() && self.consumer_verified.is_none() => {
self.consumer_verified = Some(observed_policy_revision.clone());
}
DevelopmentReconcileTransition::MaterialPublished { file, .. }
if self.consumer_verified.is_some() && self.material_published.is_none() =>
{
self.material_published = Some(file.clone());
}
DevelopmentReconcileTransition::Terminal { outcome, .. } if self.effect_deadline.is_some() => {
self.terminal = Some(*outcome);
}
DevelopmentReconcileTransition::Completed if self.material_published.is_some() => {
self.completed = true;
}
_ => return Err(WorkflowJournalError::Corrupt),
}
Ok(())
}
fn role_target_is_next(&self, target: DevelopmentReconcileRoleTarget) -> bool {
self.index_active
&& match target {
DevelopmentReconcileRoleTarget::Administrator => !self
.role_verified
.contains_key(&DevelopmentReconcileRoleTarget::Administrator),
DevelopmentReconcileRoleTarget::Consumer => {
self.role_verified
.contains_key(&DevelopmentReconcileRoleTarget::Administrator)
&& !self
.role_verified
.contains_key(&DevelopmentReconcileRoleTarget::Consumer)
}
}
}
fn role_verification_matches(
&self,
target: DevelopmentReconcileRoleTarget,
role_version: &U64,
observed_policy_revision: &U64,
) -> bool {
match self.role_preparations.get(&target) {
Some((_, _, _, _, successor_role_version)) => {
successor_role_version == role_version
&& self.role_applied.get(&target).is_some_and(|target_policy_revision| {
ordered_or_equal_u64(target_policy_revision, observed_policy_revision)
})
}
None => !self.role_applied.contains_key(&target),
}
}
#[must_use]
pub const fn effect_deadline(&self) -> Option<&U64> {
self.effect_deadline.as_ref()
}
#[must_use]
pub const fn resource(&self) -> Option<&Resource> {
self.resource.as_ref()
}
#[must_use]
pub const fn authority_binding(&self) -> Option<&(Identity, Identity, Digest, U64, U64, U64)> {
self.binding.as_ref()
}
#[must_use]
pub const fn catalog_preparation(&self) -> Option<&(Locator, Digest, U64, IndexResourceSpec)> {
self.catalog_preparation.as_ref()
}
#[must_use]
pub const fn index_active(&self) -> bool {
self.index_active
}
#[must_use]
pub fn role_preparation(
&self,
target: DevelopmentReconcileRoleTarget,
) -> Option<&(Locator, Digest, U64, U64, U64)> {
self.role_preparations.get(&target)
}
#[must_use]
pub fn role_applied(&self, target: DevelopmentReconcileRoleTarget) -> Option<&U64> {
self.role_applied.get(&target)
}
#[must_use]
pub fn role_verified(&self, target: DevelopmentReconcileRoleTarget) -> Option<&(U64, U64)> {
self.role_verified.get(&target)
}
#[must_use]
pub const fn reservation(&self) -> Option<&ServiceCredentialReservation> {
self.reservation.as_ref()
}
#[must_use]
pub const fn credential_staged(&self) -> Option<&PublishedFileIdentity> {
self.credential_staged.as_ref()
}
#[must_use]
pub const fn credential_activated(&self) -> bool {
self.credential_activated
}
#[must_use]
pub const fn credential_published(&self) -> Option<&PublishedFileIdentity> {
self.credential_published.as_ref()
}
#[must_use]
pub const fn profile_published(&self) -> Option<&PublishedFileIdentity> {
self.profile_published.as_ref()
}
#[must_use]
pub const fn consumer_policy_revision(&self) -> Option<&U64> {
self.consumer_verified.as_ref()
}
#[must_use]
pub const fn material_published(&self) -> Option<&PublishedFileIdentity> {
self.material_published.as_ref()
}
#[must_use]
pub const fn terminal(&self) -> Option<DevelopmentReconcileTerminalOutcome> {
self.terminal
}
#[must_use]
pub const fn completed(&self) -> bool {
self.completed
}
}
fn valid_absolute_path(value: &str) -> bool {
if value.is_empty() || value.len() > 4_096 || value.bytes().any(|byte| byte.is_ascii_control()) {
return false;
}
let path = Path::new(value);
path.is_absolute()
&& path.parent().is_some()
&& path
.components()
.all(|component| matches!(component, Component::RootDir | Component::Normal(_)))
&& path.components().collect::<PathBuf>().as_os_str() == path.as_os_str()
}
fn valid_sha256(value: &[u8; 32]) -> bool {
value.iter().any(|byte| *byte != 0)
}
fn valid_instance_identity(value: &str) -> bool {
value.len() == 64
&& value
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
&& value.bytes().any(|byte| byte != b'0')
}
fn valid_git_object(value: &str) -> bool {
matches!(value.len(), 40 | 64)
&& value
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
&& value.bytes().any(|byte| byte != b'0')
}
fn valid_development_endpoints(native: &str, subscription: &str, internal: &str, bootstrap: &str) -> bool {
let values = [native, subscription, internal, bootstrap];
if values.iter().any(|value| {
value.is_empty() || value.len() > MAX_ENDPOINT_BYTES || !value.bytes().all(|byte| byte.is_ascii_graphic())
}) {
return false;
}
let Some(authority) = native.strip_prefix("lodestone+tls://") else {
return false;
};
let (advertised_host, port_text) = if let Some(port) = authority.strip_prefix("[::1]:") {
("[::1]", port)
} else if let Some(port) = authority.strip_prefix("localhost:") {
("localhost", port)
} else if let Some(port) = authority.strip_prefix("127.0.0.1:") {
("127.0.0.1", port)
} else {
return false;
};
let Some(base_port) = port_text
.parse::<u16>()
.ok()
.filter(|port| port.to_string() == port_text && (1024..=65532).contains(port))
else {
return false;
};
let bind_host = match advertised_host {
"[::1]" => "[::1]",
_ => "127.0.0.1",
};
subscription == format!("https://{advertised_host}:{}", base_port + 1)
&& internal == format!("{bind_host}:{}", base_port + 2)
&& bootstrap == format!("{bind_host}:{}", base_port + 3)
}
fn valid_identity(value: &Identity) -> bool {
value.value.len() == 32
&& value
.value
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
&& value.value.bytes().any(|byte| byte != b'0')
}
fn valid_u64(value: &U64) -> bool {
value
.value
.parse::<u64>()
.ok()
.is_some_and(|parsed| parsed.to_string() == value.value)
}
fn canonical_u64(value: &U64) -> Option<u64> {
value
.value
.parse::<u64>()
.ok()
.filter(|parsed| parsed.to_string() == value.value)
}
fn ordered_successor(predecessor: &U64, successor: &U64) -> bool {
canonical_u64(predecessor)
.filter(|value| *value > 0)
.and_then(|value| value.checked_add(1))
.is_some_and(|expected| canonical_u64(successor) == Some(expected))
}
fn ordered_u64(predecessor: &U64, successor: &U64) -> bool {
canonical_u64(predecessor)
.zip(canonical_u64(successor))
.is_some_and(|(predecessor, successor)| predecessor < successor)
}
fn ordered_or_equal_u64(predecessor: &U64, successor: &U64) -> bool {
canonical_u64(predecessor)
.zip(canonical_u64(successor))
.is_some_and(|(predecessor, successor)| predecessor <= successor)
}
fn checked_add_u64(left: &U64, right: &U64) -> Option<u64> {
canonical_u64(left)?.checked_add(canonical_u64(right)?)
}
fn valid_positive_bounded_duration(value: &U64) -> bool {
canonical_u64(value).is_some_and(|value| (1..=86_400_000).contains(&value))
}
fn valid_digest(value: &Digest) -> bool {
value.value.len() == 64
&& value
.value
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
&& value.value.bytes().any(|byte| byte != b'0')
}
fn valid_locator(value: &Locator) -> bool {
!value.authority.is_empty()
&& value.authority.len() <= MAX_NAME_BYTES
&& value.authority.bytes().all(|byte| byte.is_ascii_graphic())
&& !value.value.is_empty()
&& value.value.len() <= 4_096
&& value.value.bytes().all(|byte| byte.is_ascii_graphic())
}
fn valid_resource(value: &Resource) -> bool {
value.kind == "index" && valid_index_name(&value.name) && valid_identity(&value.incarnation)
}
fn valid_index_name(value: &str) -> bool {
!value.is_empty()
&& value.len() <= MAX_NAME_BYTES
&& value
.bytes()
.all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'-' | b'_'))
}
fn valid_reservation(value: &ServiceCredentialReservation) -> bool {
valid_locator(&value.locator)
&& value.locator.authority == "service_credential_reservation"
&& valid_digest(&value.reservation_action_id)
&& valid_digest(&value.activation_action_id)
&& valid_identity(&value.principal_id)
&& valid_identity(&value.api_key_id)
&& valid_u64(&value.generation)
&& valid_u64(&value.reservation_expires_at)
&& valid_u64(&value.credential_expires_at)
&& value.verifier_algorithm == "sha256_identity_bound_v1"
&& value.verifier_algorithm_version == 1
&& value.required_secret_bytes == 32
}
#[cfg(test)]
#[allow(clippy::expect_used)] mod tests {
use super::*;
fn identity(value: u128) -> Identity {
Identity {
value: format!("{value:032x}"),
}
}
fn instance_identity(value: u8) -> String {
format!("{value:02x}").repeat(32)
}
fn digest(value: u8) -> Digest {
Digest {
value: format!("{value:02x}").repeat(32),
}
}
fn u64_value(value: u64) -> U64 {
U64 {
value: value.to_string(),
}
}
fn file(value: u64) -> PublishedFileIdentity {
PublishedFileIdentity {
device: 1,
inode: value,
size: 1,
sha256: [u8::try_from(value).unwrap_or(1); 32],
}
}
fn reconcile_intent() -> DevelopmentReconcileIntent {
DevelopmentReconcileIntent {
format_version: DEVELOPMENT_RECONCILE_FORMAT_VERSION,
instance_root: "/tmp/lodestone".to_owned(),
instance_manifest_sha256: [1; 32],
index_definition_path: "/tmp/lodestone/operator/development-index.json".to_owned(),
index_definition_sha256: [2; 32],
credential_output: "/tmp/output/credential.json".to_owned(),
profile_output: "/tmp/output/profile.json".to_owned(),
material_output: "/tmp/output/material.json".to_owned(),
timeout_millis: u64_value(90),
reservation_ttl_millis: u64_value(10),
credential_ttl_millis: u64_value(20),
}
}
#[test]
fn materialization_manifest_is_last_and_completion_requires_verification() -> Result<(), WorkflowJournalError> {
let mut transitions = vec![
DevelopmentMaterializeTransition::InstanceIdentityRecorded {
instance_identity: instance_identity(1),
},
DevelopmentMaterializeTransition::ClockRecorded {
unix_millis: u64_value(1),
},
DevelopmentMaterializeTransition::PrivatePlanStaged {
plan: DevelopmentMaterializePrivatePlan::Pki,
file: file(200),
},
DevelopmentMaterializeTransition::PrivatePlanPublished {
plan: DevelopmentMaterializePrivatePlan::Pki,
file: file(200),
},
];
for (position, artifact) in DevelopmentMaterializeArtifact::required_before_manifest()
.into_iter()
.enumerate()
{
let file = file(u64::try_from(position).unwrap_or(0) + 1);
transitions.push(DevelopmentMaterializeTransition::ArtifactStaged {
artifact,
file: file.clone(),
});
transitions.push(DevelopmentMaterializeTransition::ArtifactPublished {
artifact,
file,
public_sha256: None,
});
}
let manifest = file(100);
transitions.push(DevelopmentMaterializeTransition::ArtifactStaged {
artifact: DevelopmentMaterializeArtifact::InstanceManifest,
file: manifest.clone(),
});
transitions.push(DevelopmentMaterializeTransition::ArtifactPublished {
artifact: DevelopmentMaterializeArtifact::InstanceManifest,
file: manifest,
public_sha256: Some([3; 32]),
});
assert!(DevelopmentMaterializeState::replay(&transitions).is_ok());
transitions.push(DevelopmentMaterializeTransition::Verified {
manifest_sha256: [3; 32],
});
transitions.push(DevelopmentMaterializeTransition::Completed);
assert!(DevelopmentMaterializeState::replay(&transitions)?.completed());
let out_of_order = [
DevelopmentMaterializeTransition::InstanceIdentityRecorded {
instance_identity: instance_identity(1),
},
DevelopmentMaterializeTransition::ClockRecorded {
unix_millis: u64_value(1),
},
DevelopmentMaterializeTransition::ArtifactPublished {
artifact: DevelopmentMaterializeArtifact::InstanceManifest,
file: file(1),
public_sha256: Some([3; 32]),
},
];
assert_eq!(
DevelopmentMaterializeState::replay(&out_of_order),
Err(WorkflowJournalError::Corrupt)
);
Ok(())
}
#[test]
fn pki_successor_requires_ca_artifacts_only_for_ca_rotation() -> Result<(), WorkflowJournalError> {
let intent = DevelopmentPkiSuccessorIntent {
format_version: DEVELOPMENT_PKI_SUCCESSOR_FORMAT_VERSION,
operation: DevelopmentPkiSuccessorOperation::RenewLeaves,
instance_identity: instance_identity(1),
parent_manifest_path: "/tmp/parent/instance.json".to_owned(),
parent_manifest_sha256: [1; 32],
successor_root: "/tmp/successor".to_owned(),
parent_instance_generation: u64_value(1),
successor_instance_generation: u64_value(2),
parent_ca_generation: u64_value(1),
successor_ca_generation: u64_value(1),
parent_process_leaf_generation: u64_value(1),
successor_process_leaf_generation: u64_value(2),
parent_gateway_leaf_generation: u64_value(1),
successor_gateway_leaf_generation: u64_value(2),
};
intent.validate()?;
let illegal = [DevelopmentPkiSuccessorTransition::ArtifactStaged {
artifact: DevelopmentPkiSuccessorArtifact::CaPrivateKey,
file: file(1),
}];
assert_eq!(
DevelopmentPkiSuccessorState::replay(&intent, &illegal),
Err(WorkflowJournalError::Corrupt)
);
let mut transitions = vec![
DevelopmentPkiSuccessorTransition::ClockRecorded {
unix_millis: u64_value(1),
},
DevelopmentPkiSuccessorTransition::PrivatePlanStaged {
plan: DevelopmentPkiSuccessorPrivatePlan::Pki,
file: file(90),
},
DevelopmentPkiSuccessorTransition::PrivatePlanPublished {
plan: DevelopmentPkiSuccessorPrivatePlan::Pki,
file: file(90),
},
];
for (position, artifact) in required_pki_artifacts(intent.operation).into_iter().enumerate() {
let identity = file(u64::try_from(position).unwrap_or(0) + 1);
transitions.push(DevelopmentPkiSuccessorTransition::ArtifactStaged {
artifact,
file: identity.clone(),
});
transitions.push(DevelopmentPkiSuccessorTransition::ArtifactPublished {
artifact,
file: identity,
public_sha256: None,
});
}
let manifest = file(100);
transitions.push(DevelopmentPkiSuccessorTransition::ArtifactStaged {
artifact: DevelopmentPkiSuccessorArtifact::InstanceManifest,
file: manifest.clone(),
});
transitions.push(DevelopmentPkiSuccessorTransition::ArtifactPublished {
artifact: DevelopmentPkiSuccessorArtifact::InstanceManifest,
file: manifest,
public_sha256: Some([9; 32]),
});
transitions.push(DevelopmentPkiSuccessorTransition::Verified {
manifest_sha256: [9; 32],
});
transitions.push(DevelopmentPkiSuccessorTransition::Completed);
assert!(DevelopmentPkiSuccessorState::replay(&intent, &transitions)?.completed());
Ok(())
}
#[test]
fn reconciliation_enforces_administrator_scope_before_consumer_scope() -> Result<(), WorkflowJournalError> {
let resource = Resource {
kind: "index".to_owned(),
name: "events".to_owned(),
incarnation: identity(4),
};
let prefix = [
DevelopmentReconcileTransition::EffectDeadlineRecorded {
unix_millis: u64_value(100),
},
DevelopmentReconcileTransition::AuthorityObserved {
cluster_id: identity(1),
collection_id: identity(2),
schema_sha256: digest(3),
logical_time: u64_value(10),
reservation_expires_at: u64_value(20),
credential_expires_at: u64_value(30),
},
DevelopmentReconcileTransition::IndexActive {
resource,
index_revision: u64_value(1),
},
DevelopmentReconcileTransition::RoleVerified {
target: DevelopmentReconcileRoleTarget::Consumer,
role_version: u64_value(1),
observed_policy_revision: u64_value(1),
},
];
assert_eq!(
DevelopmentReconcileState::replay(&reconcile_intent(), &prefix),
Err(WorkflowJournalError::Corrupt)
);
Ok(())
}
#[test]
fn reconciliation_terminal_outcome_is_final_but_unresolved_work_remains_resumable() {
let deadline = DevelopmentReconcileTransition::EffectDeadlineRecorded {
unix_millis: u64_value(100),
};
let terminal = DevelopmentReconcileTransition::Terminal {
outcome: DevelopmentReconcileTerminalOutcome::EffectDeadlineExpired,
};
let state = DevelopmentReconcileState::replay(&reconcile_intent(), &[deadline.clone(), terminal]);
assert_eq!(
state.map(|value| value.terminal()),
Ok(Some(DevelopmentReconcileTerminalOutcome::EffectDeadlineExpired))
);
assert_eq!(
DevelopmentReconcileState::replay(
&reconcile_intent(),
&[
deadline,
DevelopmentReconcileTransition::Terminal {
outcome: DevelopmentReconcileTerminalOutcome::Conflict,
},
DevelopmentReconcileTransition::Completed,
],
),
Err(WorkflowJournalError::Corrupt)
);
let mut resumable = vec![
DevelopmentReconcileTransition::EffectDeadlineRecorded {
unix_millis: u64_value(100),
},
DevelopmentReconcileTransition::AuthorityObserved {
cluster_id: identity(1),
collection_id: identity(2),
schema_sha256: digest(3),
logical_time: u64_value(10),
reservation_expires_at: u64_value(20),
credential_expires_at: u64_value(30),
},
DevelopmentReconcileTransition::CatalogPrepared {
locator: Locator {
authority: "catalog".to_owned(),
value: "create-events".to_owned(),
},
semantic_digest: digest(4),
expires_at: u64_value(40),
resource_spec: IndexResourceSpec {
format_version: u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION),
revision: u64_value(1),
initial_shards: 1,
target_shard_bytes: u64_value(1),
durability: crate::ResourceDurabilityClass::RegionalObject,
storage_class: crate::ResourceStorageClass::Standard,
ram_admission_weight: 1,
ram_hard_max_bytes: u64_value(1),
nvme_admission_weight: 1,
nvme_hard_max_bytes: u64_value(1),
refresh_objective_millis: 1,
memtable_hard_max_bytes: u64_value(1),
seal_parallelism: 1,
merge_parallelism: 1,
vector: lodestone_wire::VectorResourcePolicy {
quality: lodestone_wire::VectorQualityObjective::Balanced,
indexing: lodestone_wire::VectorIndexingObjective::Balanced,
residency: None,
calibration: None,
},
lifecycle_binding: None,
encryption_required: false,
resolver_capability: None,
},
},
];
let mut replacement = resumable[2].clone();
if let DevelopmentReconcileTransition::CatalogPrepared {
locator,
semantic_digest,
..
} = &mut replacement
{
locator.value = "create-events-after-root-movement".to_owned();
*semantic_digest = digest(5);
}
resumable.push(replacement);
let state = DevelopmentReconcileState::replay(&reconcile_intent(), &resumable);
assert!(state.is_ok_and(|value| {
value.terminal().is_none()
&& value
.catalog_preparation()
.is_some_and(|(locator, semantic_digest, _, _)| {
locator.value == "create-events-after-root-movement" && semantic_digest == &digest(5)
})
}));
}
}