use std::{
future::Future,
sync::atomic::{AtomicBool, Ordering},
};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::c5_compiled::{CompiledDbPermit, CompiledExecutionWithCapacityLeaf, ExecutionError};
use saddle_admission::{ManagedBytes, ManagedResponse, ManagedResponseBuilder};
const SCHEMA: &str = "saddle-0.2-service-production-fact/1";
const DOMAIN: &str = "service";
const MANIFEST_LEAF: &str = "service";
const COMMITMENT: &str = "service_capacity";
const APPROVED_SOURCE: &str = "da64568f31bfacd38f6c0807588dd2cc59c258fe";
const APPROVED_WHOLE: &[u8] = include_bytes!("approved-inputs/candidate-fact-whole.json");
const APPROVED_PERMIT: &[u8] = include_bytes!("approved-inputs/permit.json");
static SERVICE_CANDIDATE_CONSUMED: AtomicBool = AtomicBool::new(false);
#[doc(hidden)]
pub struct ServiceProductionFactInput {
document: ServiceProductionFactDocument,
}
#[doc(hidden)]
pub struct ServiceSourceCandidateInput {
whole: &'static [u8],
permit: &'static [u8],
}
#[doc(hidden)]
pub struct VerifiedServiceSourceCandidateOwner {
fact: ServiceProductionFactInput,
candidate: ServiceSourceCandidateInput,
}
impl VerifiedServiceSourceCandidateOwner {
#[doc(hidden)]
pub fn rollback(self) -> (ServiceProductionFactInput, ServiceSourceCandidateInput) {
SERVICE_CANDIDATE_CONSUMED.store(false, Ordering::Release);
(self.fact, self.candidate)
}
}
#[doc(hidden)]
pub struct ServiceSourceCandidateRejection {
fact: ServiceProductionFactInput,
candidate: ServiceSourceCandidateInput,
}
impl ServiceSourceCandidateRejection {
#[doc(hidden)]
pub fn into_inputs(self) -> (ServiceProductionFactInput, ServiceSourceCandidateInput) {
(self.fact, self.candidate)
}
}
#[derive(Debug)]
#[doc(hidden)]
pub enum ServiceProductionFactError {
RouteClosure,
Serialization(serde_json::Error),
}
impl core::fmt::Display for ServiceProductionFactError {
fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::RouteClosure => formatter.write_str("Service production route closure failed"),
Self::Serialization(_) => {
formatter.write_str("Service production fact serialization failed")
}
}
}
}
impl std::error::Error for ServiceProductionFactError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::RouteClosure => None,
Self::Serialization(error) => Some(error),
}
}
}
impl ServiceProductionFactInput {
#[doc(hidden)]
pub fn into_canonical_json(self) -> Result<Box<[u8]>, ServiceProductionFactError> {
serde_json::to_vec(&self.document)
.map(Vec::into_boxed_slice)
.map_err(ServiceProductionFactError::Serialization)
}
}
#[doc(hidden)]
pub fn service_source_candidate_input() -> ServiceSourceCandidateInput {
ServiceSourceCandidateInput {
whole: APPROVED_WHOLE,
permit: APPROVED_PERMIT,
}
}
#[doc(hidden)]
#[allow(clippy::result_large_err)]
pub fn bind_service_source_candidate(
fact: ServiceProductionFactInput,
candidate: ServiceSourceCandidateInput,
) -> Result<VerifiedServiceSourceCandidateOwner, ServiceSourceCandidateRejection> {
if !matches_approved_candidate(&fact, &candidate)
|| SERVICE_CANDIDATE_CONSUMED
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
return Err(ServiceSourceCandidateRejection { fact, candidate });
}
Ok(VerifiedServiceSourceCandidateOwner { fact, candidate })
}
fn matches_approved_candidate(
fact: &ServiceProductionFactInput,
candidate: &ServiceSourceCandidateInput,
) -> bool {
if candidate.whole.as_ptr() != APPROVED_WHOLE.as_ptr()
|| candidate.whole.len() != APPROVED_WHOLE.len()
|| candidate.permit.as_ptr() != APPROVED_PERMIT.as_ptr()
|| candidate.permit.len() != APPROVED_PERMIT.len()
{
return false;
}
let Ok(fact_bytes) = serde_json::to_vec(&fact.document) else {
return false;
};
let Ok(whole) = serde_json::from_slice::<CandidateWhole>(candidate.whole) else {
return false;
};
let Ok(permit) = serde_json::from_slice::<ValidationPermit>(candidate.permit) else {
return false;
};
let mut route_identities: Vec<&str> = fact
.document
.capacity
.routes
.iter()
.map(|route| route.route_identity.as_str())
.collect();
route_identities.sort_unstable();
let mut approved_route_identities: Vec<&str> = whole
.projection
.route_identities
.iter()
.map(String::as_str)
.collect();
approved_route_identities.sort_unstable();
whole.schema == "saddle-0.2-semantic-fact-whole-candidate/1"
&& !whole.authority
&& whole.projection.domain_sha256.service == hex_sha256(&fact_bytes)
&& approved_route_identities == route_identities
&& permit.schema == "saddle-0.2-golden-c8-source-validation-permit/1"
&& permit.authority_scope == "golden-c8-listener-preclosure-only"
&& permit.source_candidate_identity == APPROVED_SOURCE
&& permit.candidate_fact_whole_identity == hex_sha256(candidate.whole)
&& permit.candidate_semantic_identity == whole.semantic_identity
&& permit.single_use
&& permit.minimum_terminal_stage == "listener"
&& !permit.signing_authority
&& !permit.enterprise_production_authority
&& !permit.rust_skill_artifact_combination_authority
&& !permit.component_production_wiring_authority
&& !permit.publish_authority
&& !permit.release_authority
}
pub(super) fn service_production_fact_input<E, C, F, const BODY: usize, const OUTPUT: usize>(
bundle: &CompiledExecutionWithCapacityLeaf<E, C, F, BODY, OUTPUT>,
) -> Result<ServiceProductionFactInput, ServiceProductionFactError>
where
E: Fn(usize, C, ManagedBytes, CompiledDbPermit, ManagedResponseBuilder) -> F
+ Send
+ Sync
+ 'static,
C: Send + Unpin + 'static,
F: Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static,
{
let mut routes = Vec::new();
routes
.try_reserve_exact(bundle.service_leaf.routes.len())
.map_err(|_| ServiceProductionFactError::RouteClosure)?;
for route in bundle.execution.entries.canonical_route_facts() {
let identity = route.identity.contract().opaque();
let capacity = bundle
.service_leaf
.routes
.iter()
.find(|candidate| candidate.identity == identity)
.ok_or(ServiceProductionFactError::RouteClosure)?;
let token = std::str::from_utf8(route.token)
.map_err(|_| ServiceProductionFactError::RouteClosure)?;
routes.push(ServiceRouteFact {
route: token.to_owned(),
framing: route.framing.canonical_name().to_owned(),
route_identity: hex_u64(identity),
route_source_identity: hex_identity(capacity.source_identity),
managed_commitment_bytes: capacity.commitment,
managed_objects_peak: capacity.managed_objects_peak,
db_connections: capacity.db_connections,
db_operations: capacity.db_operations,
});
}
if routes.len() != bundle.service_leaf.routes.len() {
return Err(ServiceProductionFactError::RouteClosure);
}
Ok(ServiceProductionFactInput {
document: ServiceProductionFactDocument {
schema: SCHEMA.to_owned(),
domain: DOMAIN.to_owned(),
authority: false,
capacity: ServiceCapacityFact {
manifest_leaf: MANIFEST_LEAF.to_owned(),
commitment: COMMITMENT.to_owned(),
leaf_identity: hex_identity(bundle.service_leaf.leaf_identity),
route_set_identity: hex_identity(bundle.service_leaf.common_identities[2]),
route_type_closure_identity: hex_identity(
bundle.service_leaf.route_type_closure_identity,
),
routes,
},
},
})
}
#[derive(Deserialize, Serialize)]
struct ServiceProductionFactDocument {
schema: String,
domain: String,
authority: bool,
capacity: ServiceCapacityFact,
}
#[derive(Deserialize, Serialize)]
struct ServiceCapacityFact {
manifest_leaf: String,
commitment: String,
leaf_identity: String,
route_set_identity: String,
route_type_closure_identity: String,
routes: Vec<ServiceRouteFact>,
}
#[derive(Deserialize, Serialize)]
struct ServiceRouteFact {
route: String,
framing: String,
route_identity: String,
route_source_identity: String,
managed_commitment_bytes: usize,
managed_objects_peak: usize,
db_connections: usize,
db_operations: usize,
}
fn hex_identity(identity: [u8; 32]) -> String {
use core::fmt::Write as _;
let mut encoded = String::with_capacity(64);
for byte in identity {
write!(&mut encoded, "{byte:02x}").expect("writing to String cannot fail");
}
encoded
}
fn hex_u64(identity: u64) -> String {
format!("{identity:016x}")
}
fn hex_sha256(bytes: &[u8]) -> String {
hex_identity(Sha256::digest(bytes).into())
}
#[derive(Deserialize)]
struct CandidateWhole {
schema: String,
authority: bool,
semantic_identity: String,
projection: CandidateProjection,
}
#[derive(Deserialize)]
struct CandidateProjection {
domain_sha256: CandidateDomainDigests,
route_identities: Vec<String>,
}
#[derive(Deserialize)]
struct CandidateDomainDigests {
service: String,
}
#[derive(Deserialize)]
struct ValidationPermit {
schema: String,
authority_scope: String,
source_candidate_identity: String,
candidate_fact_whole_identity: String,
candidate_semantic_identity: String,
single_use: bool,
minimum_terminal_stage: String,
signing_authority: bool,
enterprise_production_authority: bool,
rust_skill_artifact_combination_authority: bool,
component_production_wiring_authority: bool,
publish_authority: bool,
release_authority: bool,
}
#[cfg(test)]
mod candidate_tests {
use super::*;
fn approved_fact() -> ServiceProductionFactInput {
let machine: serde_json::Value = serde_json::from_slice(include_bytes!(
"approved-inputs/canonical-machine-input.json"
))
.expect("approved machine input parses");
let document = serde_json::from_value(machine["facts"]["service"].clone())
.expect("approved Service fact shape parses");
ServiceProductionFactInput { document }
}
#[test]
fn approved_candidate_is_exact_recoverable_and_retryable_after_rollback() {
let foreign_whole = Box::leak(APPROVED_WHOLE.to_vec().into_boxed_slice());
let foreign = ServiceSourceCandidateInput {
whole: foreign_whole,
permit: APPROVED_PERMIT,
};
let foreign_whole_pointer = foreign.whole.as_ptr();
let foreign_permit_pointer = foreign.permit.as_ptr();
let rejected = bind_service_source_candidate(approved_fact(), foreign)
.err()
.expect("foreign whole rejects");
let (_fact, foreign) = rejected.into_inputs();
assert_eq!(foreign.whole.as_ptr(), foreign_whole_pointer);
assert_eq!(foreign.permit.as_ptr(), foreign_permit_pointer);
let foreign_permit = Box::leak(APPROVED_PERMIT.to_vec().into_boxed_slice());
let foreign = ServiceSourceCandidateInput {
whole: APPROVED_WHOLE,
permit: foreign_permit,
};
let foreign_whole_pointer = foreign.whole.as_ptr();
let foreign_permit_pointer = foreign.permit.as_ptr();
let rejected = bind_service_source_candidate(approved_fact(), foreign)
.err()
.expect("foreign permit rejects");
let (fact, foreign) = rejected.into_inputs();
assert_eq!(foreign.whole.as_ptr(), foreign_whole_pointer);
assert_eq!(foreign.permit.as_ptr(), foreign_permit_pointer);
let mut drifted = approved_fact();
drifted.document.capacity.routes[0].managed_commitment_bytes += 1;
let drift_candidate = service_source_candidate_input();
let drift_whole = drift_candidate.whole.as_ptr();
let drift_permit = drift_candidate.permit.as_ptr();
let rejected = bind_service_source_candidate(drifted, drift_candidate)
.err()
.expect("drifted Service fact rejects");
let (drifted, drift_candidate) = rejected.into_inputs();
assert_eq!(
drifted.document.capacity.routes[0].managed_commitment_bytes,
approved_fact().document.capacity.routes[0].managed_commitment_bytes + 1
);
assert_eq!(drift_candidate.whole.as_ptr(), drift_whole);
assert_eq!(drift_candidate.permit.as_ptr(), drift_permit);
let candidate = service_source_candidate_input();
let fact_bytes = serde_json::to_vec(&fact.document).unwrap();
let whole: CandidateWhole = serde_json::from_slice(candidate.whole).unwrap();
let permit: ValidationPermit = serde_json::from_slice(candidate.permit).unwrap();
assert_eq!(
whole.projection.domain_sha256.service,
hex_sha256(&fact_bytes)
);
assert_eq!(permit.source_candidate_identity, APPROVED_SOURCE);
assert_eq!(
permit.candidate_fact_whole_identity,
hex_sha256(candidate.whole)
);
assert_eq!(permit.candidate_semantic_identity, whole.semantic_identity);
assert!(matches_approved_candidate(&fact, &candidate));
let owner = bind_service_source_candidate(fact, candidate)
.unwrap_or_else(|_| panic!("approved pair binds"));
let replay_candidate = service_source_candidate_input();
let replay_whole = replay_candidate.whole.as_ptr();
let replay_permit = replay_candidate.permit.as_ptr();
let rejected = bind_service_source_candidate(approved_fact(), replay_candidate)
.err()
.expect("approved pair cannot replay");
let (replay_fact, replay_candidate) = rejected.into_inputs();
assert_eq!(
hex_sha256(&serde_json::to_vec(&replay_fact.document).unwrap()),
whole.projection.domain_sha256.service
);
assert_eq!(replay_candidate.whole.as_ptr(), replay_whole);
assert_eq!(replay_candidate.permit.as_ptr(), replay_permit);
let (fact, candidate) = owner.rollback();
let retried = bind_service_source_candidate(fact, candidate)
.unwrap_or_else(|_| panic!("rolled-back pair retries"));
let _restored = retried.rollback();
}
}