use std::collections::{BTreeMap, BTreeSet};
use std::error::Error;
use std::fmt;
use serde_json::json;
use sha2::{Digest, Sha256};
use crate::{ChunkSize, JobName, StateSchemaId, StateSchemaVersion, StepName};
const MAX_TOKEN_BYTES: usize = 128;
pub const MAX_NODES: usize = 1_024;
pub const MAX_TRANSITIONS: usize = 4_096;
pub(crate) const MAX_MANIFEST_BYTES: usize = 64 * 1024;
pub const MANIFEST_FORMAT_ONE_STEP: u16 = 1;
pub const MANIFEST_FORMAT_FLOW: u16 = 2;
pub const MANIFEST_FORMAT_LOCAL_SCALE: u16 = 3;
pub(crate) const SUPPORTED_MANIFEST_FORMAT: u16 = MANIFEST_FORMAT_LOCAL_SCALE;
const LEGACY_REVISION: &str = "__m1_repository_port_v1";
const LEGACY_MANIFEST: &[u8] =
br#"{"format":1,"repository_port":"m1","revision":"__m1_repository_port_v1"}"#;
pub fn validate_token(value: &str, kind: DefinitionTokenKind) -> Result<(), DefinitionError> {
if value.is_empty() {
return Err(DefinitionError::EmptyToken { kind });
}
if value.len() > MAX_TOKEN_BYTES {
return Err(DefinitionError::TokenTooLong {
kind,
max_bytes: MAX_TOKEN_BYTES,
});
}
if value.trim() != value {
return Err(DefinitionError::SurroundingWhitespace { kind });
}
if value.chars().any(char::is_control) {
return Err(DefinitionError::ControlCharacter { kind });
}
Ok(())
}
#[macro_export]
macro_rules! definition_token {
($name:ident, $kind:expr, $docs:literal) => {
#[doc = $docs]
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct $name(String);
impl $name {
pub fn new(value: impl Into<String>) -> Result<Self, DefinitionError> {
let value = value.into();
validate_token(&value, $kind)?;
Ok(Self(value))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
};
}
definition_token!(
DefinitionRevision,
DefinitionTokenKind::Revision,
"An application-owned audit label for one restart-relevant definition."
);
definition_token!(
DefinitionUpgradeKey,
DefinitionTokenKind::Upgrade,
"An application-owned key for one directed definition compatibility edge."
);
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StepDefinitionUpgrade {
source: StepName,
target: StepName,
}
impl StepDefinitionUpgrade {
#[must_use]
pub const fn new(source: StepName, target: StepName) -> Self {
Self { source, target }
}
#[must_use]
pub const fn source(&self) -> &StepName {
&self.source
}
#[must_use]
pub const fn target(&self) -> &StepName {
&self.target
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DefinitionUpgrade {
key: DefinitionUpgradeKey,
from: DefinitionIdentity,
to: DefinitionIdentity,
step_mapping: BTreeMap<StepName, StepName>,
}
impl DefinitionUpgrade {
pub fn new(
key: DefinitionUpgradeKey,
from: DefinitionIdentity,
to: DefinitionIdentity,
steps: impl IntoIterator<Item = StepDefinitionUpgrade>,
) -> Result<Self, DefinitionError> {
if from.manifest_digest() == to.manifest_digest() {
return Err(DefinitionError::UpgradeSelfEdge);
}
let mut step_mapping = BTreeMap::new();
let mut targets = BTreeSet::new();
for step in steps {
if step_mapping
.insert(step.source().clone(), step.target().clone())
.is_some()
{
return Err(DefinitionError::DuplicateSourceStep);
}
if !targets.insert(step.target().clone()) {
return Err(DefinitionError::DuplicateTargetStep);
}
}
if step_mapping.is_empty() {
return Err(DefinitionError::EmptyStepMapping);
}
Ok(Self {
key,
from,
to,
step_mapping,
})
}
#[must_use]
pub const fn key(&self) -> &DefinitionUpgradeKey {
&self.key
}
#[must_use]
pub const fn from(&self) -> &DefinitionIdentity {
&self.from
}
#[must_use]
pub const fn to(&self) -> &DefinitionIdentity {
&self.to
}
#[must_use]
pub fn step_mapping(&self) -> &BTreeMap<StepName, StepName> {
&self.step_mapping
}
}
definition_token!(
ComponentRevision,
DefinitionTokenKind::Component,
"An application-owned revision token for one opaque executable component."
);
definition_token!(
ClassifierRevision,
DefinitionTokenKind::Classifier,
"An application-owned revision token for one bounded fault classifier."
);
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ChunkComponentRevisions {
reader: ComponentRevision,
processor: ComponentRevision,
writer: ComponentRevision,
checkpoint: ComponentRevision,
restart: ChunkRestartContract,
}
impl ChunkComponentRevisions {
#[must_use]
pub const fn new(
reader: ComponentRevision,
processor: ComponentRevision,
writer: ComponentRevision,
checkpoint: ComponentRevision,
restart: ChunkRestartContract,
) -> Self {
Self {
reader,
processor,
writer,
checkpoint,
restart,
}
}
#[must_use]
pub const fn delivery_mode(&self) -> ChunkDeliveryMode {
self.restart.delivery_mode
}
#[must_use]
pub const fn in_flight_policy(&self) -> InFlightPolicy {
self.restart.in_flight_policy
}
#[must_use]
pub const fn reader(&self) -> &ComponentRevision {
&self.reader
}
#[must_use]
pub const fn processor(&self) -> &ComponentRevision {
&self.processor
}
#[must_use]
pub const fn writer(&self) -> &ComponentRevision {
&self.writer
}
#[must_use]
pub const fn checkpoint(&self) -> &ComponentRevision {
&self.checkpoint
}
#[must_use]
pub const fn checkpoint_schema(&self) -> &StateSchemaId {
&self.restart.checkpoint_schema
}
#[must_use]
pub const fn checkpoint_schema_version(&self) -> StateSchemaVersion {
self.restart.checkpoint_schema_version
}
#[must_use]
pub const fn context_schema(&self) -> &StateSchemaId {
&self.restart.context_schema
}
#[must_use]
pub const fn context_schema_version(&self) -> StateSchemaVersion {
self.restart.context_schema_version
}
}
#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
#[non_exhaustive]
pub enum InFlightPolicy {
#[default]
FinishChunk,
RollbackChunk,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum ChunkDeliveryMode {
AtomicSameResource,
AtLeastOnce,
}
impl ChunkDeliveryMode {
#[must_use]
pub const fn manifest_name(self) -> &'static str {
match self {
Self::AtomicSameResource => "atomic_same_resource",
Self::AtLeastOnce => "at_least_once",
}
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ChunkRestartContract {
checkpoint_schema: StateSchemaId,
checkpoint_schema_version: StateSchemaVersion,
context_schema: StateSchemaId,
context_schema_version: StateSchemaVersion,
delivery_mode: ChunkDeliveryMode,
in_flight_policy: InFlightPolicy,
}
impl ChunkRestartContract {
#[must_use]
pub const fn new(
checkpoint_schema: StateSchemaId,
checkpoint_schema_version: StateSchemaVersion,
context_schema: StateSchemaId,
context_schema_version: StateSchemaVersion,
delivery_mode: ChunkDeliveryMode,
) -> Self {
Self {
checkpoint_schema,
checkpoint_schema_version,
context_schema,
context_schema_version,
delivery_mode,
in_flight_policy: InFlightPolicy::FinishChunk,
}
}
#[must_use]
pub const fn with_in_flight_policy(mut self, policy: InFlightPolicy) -> Self {
self.in_flight_policy = policy;
self
}
}
#[derive(Clone, Eq, PartialEq)]
pub struct DefinitionIdentity {
job_name: Option<JobName>,
revision: DefinitionRevision,
manifest_format: u16,
manifest_digest: [u8; 32],
canonical_manifest: Box<[u8]>,
}
impl DefinitionIdentity {
#[must_use]
pub fn legacy() -> Self {
Self::from_canonical(
None,
DefinitionRevision(LEGACY_REVISION.to_owned()),
LEGACY_MANIFEST.to_vec(),
MANIFEST_FORMAT_ONE_STEP,
)
}
pub fn tasklet(
job_name: &JobName,
step_name: &StepName,
revision: DefinitionRevision,
component_revision: &ComponentRevision,
) -> Result<Self, DefinitionError> {
let manifest = json!({
"component": {
"tasklet": component_revision.as_str()
},
"delivery_mode": "best_effort",
"format": MANIFEST_FORMAT_ONE_STEP,
"job": job_name.as_str(),
"kind": "tasklet",
"restart_state": "none",
"step": step_name.as_str(),
"transaction_boundary": "tasklet_completion"
});
Self::encode(job_name.clone(), revision, &manifest)
}
pub fn chunk(
job_name: &JobName,
step_name: &StepName,
chunk_size: ChunkSize,
revision: DefinitionRevision,
components: &ChunkComponentRevisions,
) -> Result<Self, DefinitionError> {
let mut manifest = json!({
"chunk_size": chunk_size.get(),
"components": {
"checkpoint": components.checkpoint.as_str(),
"processor": components.processor.as_str(),
"reader": components.reader.as_str(),
"writer": components.writer.as_str()
},
"context": {
"schema": components.restart.context_schema.as_str(),
"version": components.restart.context_schema_version.get()
},
"checkpoint": {
"schema": components.restart.checkpoint_schema.as_str(),
"version": components.restart.checkpoint_schema_version.get()
},
"delivery_mode": components.restart.delivery_mode.manifest_name(),
"format": MANIFEST_FORMAT_ONE_STEP,
"job": job_name.as_str(),
"kind": "chunk",
"step": step_name.as_str(),
"transaction_boundary": "chunk"
});
if components.restart.in_flight_policy == InFlightPolicy::RollbackChunk
&& let Some(object) = manifest.as_object_mut()
{
object.insert(
"in_flight_policy".to_owned(),
serde_json::Value::String("rollback_chunk".to_owned()),
);
}
Self::encode(job_name.clone(), revision, &manifest)
}
pub fn from_flow_manifest(
job_name: &JobName,
revision: DefinitionRevision,
canonical: &[u8],
) -> Result<Self, DefinitionError> {
if canonical.len() > MAX_MANIFEST_BYTES {
return Err(DefinitionError::ManifestTooLarge {
max_bytes: MAX_MANIFEST_BYTES,
});
}
let document: serde_json::Value =
serde_json::from_slice(canonical).map_err(|_| DefinitionError::ManifestEncoding)?;
let reencoded =
serde_json::to_vec(&document).map_err(|_| DefinitionError::ManifestEncoding)?;
if !document.is_object() || reencoded != canonical {
return Err(DefinitionError::ManifestEncoding);
}
let format = document
.get("format")
.and_then(serde_json::Value::as_u64)
.and_then(|value| u16::try_from(value).ok())
.filter(|value| matches!(*value, MANIFEST_FORMAT_FLOW | MANIFEST_FORMAT_LOCAL_SCALE))
.ok_or(DefinitionError::ManifestEncoding)?;
Ok(Self::from_canonical(
Some(job_name.clone()),
revision,
canonical.to_vec(),
format,
))
}
fn encode(
job_name: JobName,
revision: DefinitionRevision,
manifest: &serde_json::Value,
) -> Result<Self, DefinitionError> {
let canonical =
serde_json::to_vec(manifest).map_err(|_| DefinitionError::ManifestEncoding)?;
if canonical.len() > MAX_MANIFEST_BYTES {
return Err(DefinitionError::ManifestTooLarge {
max_bytes: MAX_MANIFEST_BYTES,
});
}
Ok(Self::from_canonical(
Some(job_name),
revision,
canonical,
MANIFEST_FORMAT_ONE_STEP,
))
}
fn from_canonical(
job_name: Option<JobName>,
revision: DefinitionRevision,
canonical: Vec<u8>,
format: u16,
) -> Self {
let digest: [u8; 32] = Sha256::digest(&canonical).into();
Self {
job_name,
revision,
manifest_format: format,
manifest_digest: digest,
canonical_manifest: canonical.into_boxed_slice(),
}
}
#[must_use]
pub const fn revision(&self) -> &DefinitionRevision {
&self.revision
}
#[must_use]
pub const fn job_name(&self) -> Option<&JobName> {
self.job_name.as_ref()
}
#[must_use]
pub const fn manifest_format(&self) -> u16 {
self.manifest_format
}
#[must_use]
pub const fn manifest_digest(&self) -> &[u8; 32] {
&self.manifest_digest
}
#[must_use]
pub fn canonical_manifest(&self) -> &[u8] {
&self.canonical_manifest
}
}
pub const fn check_manifest_format(format: u16) -> Result<(), ManifestError> {
if format == 0 {
return Err(ManifestError::MissingFormat);
}
if format > SUPPORTED_MANIFEST_FORMAT {
return Err(ManifestError::UnsupportedFormat {
format,
supported: SUPPORTED_MANIFEST_FORMAT,
});
}
Ok(())
}
impl fmt::Debug for DefinitionIdentity {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("DefinitionIdentity")
.field("job_name", &self.job_name)
.field("revision", &self.revision)
.field("manifest_format", &self.manifest_format)
.field(
"digest_prefix",
&DigestPrefix([
self.manifest_digest[0],
self.manifest_digest[1],
self.manifest_digest[2],
self.manifest_digest[3],
]),
)
.field("canonical_manifest", &"<redacted>")
.finish()
}
}
struct DigestPrefix([u8; 4]);
impl fmt::Debug for DigestPrefix {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
for byte in self.0 {
write!(formatter, "{byte:02x}")?;
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DefinitionManifest {
format: u16,
digest: [u8; 32],
job_name: Option<JobName>,
node_count: Option<usize>,
transition_count: Option<usize>,
}
impl DefinitionManifest {
pub fn read(bytes: &[u8]) -> Result<Self, ManifestError> {
if bytes.len() > MAX_MANIFEST_BYTES {
return Err(ManifestError::TooLarge {
max_bytes: MAX_MANIFEST_BYTES,
});
}
let document: serde_json::Value =
serde_json::from_slice(bytes).map_err(|_| ManifestError::MalformedJson)?;
let members = document.as_object().ok_or(ManifestError::NotAnObject)?;
if contains_float(&document) {
return Err(ManifestError::FloatValue);
}
let reencoded = serde_json::to_vec(&document).map_err(|_| ManifestError::MalformedJson)?;
if reencoded != bytes {
return Err(ManifestError::NonCanonicalEncoding);
}
let format = members
.get("format")
.and_then(serde_json::Value::as_u64)
.and_then(|format| u16::try_from(format).ok())
.ok_or(ManifestError::MissingFormat)?;
check_manifest_format(format)?;
let job_name = members
.get("job")
.and_then(serde_json::Value::as_str)
.map(JobName::new)
.transpose()
.map_err(|_| ManifestError::InvalidJobName)?;
let (node_count, transition_count) =
if matches!(format, MANIFEST_FORMAT_FLOW | MANIFEST_FORMAT_LOCAL_SCALE) {
let nodes = array_len(members.get("nodes"))?;
let transitions = array_len(members.get("transitions"))?;
if nodes > MAX_NODES || transitions > MAX_TRANSITIONS {
return Err(ManifestError::GraphOutOfBounds {
max_nodes: MAX_NODES,
max_transitions: MAX_TRANSITIONS,
});
}
(Some(nodes), Some(transitions))
} else {
(None, None)
};
Ok(Self {
format,
digest: Sha256::digest(bytes).into(),
job_name,
node_count,
transition_count,
})
}
pub fn read_verified(bytes: &[u8], expected: &[u8; 32]) -> Result<Self, ManifestError> {
let manifest = Self::read(bytes)?;
if &manifest.digest != expected {
return Err(ManifestError::DigestMismatch);
}
Ok(manifest)
}
#[must_use]
pub const fn format(&self) -> u16 {
self.format
}
#[must_use]
pub const fn digest(&self) -> &[u8; 32] {
&self.digest
}
#[must_use]
pub const fn job_name(&self) -> Option<&JobName> {
self.job_name.as_ref()
}
#[must_use]
pub const fn node_count(&self) -> Option<usize> {
self.node_count
}
#[must_use]
pub const fn transition_count(&self) -> Option<usize> {
self.transition_count
}
}
fn array_len(value: Option<&serde_json::Value>) -> Result<usize, ManifestError> {
value
.and_then(serde_json::Value::as_array)
.map(Vec::len)
.ok_or(ManifestError::MalformedGraph)
}
fn contains_float(value: &serde_json::Value) -> bool {
match value {
serde_json::Value::Number(number) => number.as_i64().is_none() && number.as_u64().is_none(),
serde_json::Value::Array(values) => values.iter().any(contains_float),
serde_json::Value::Object(members) => members.values().any(contains_float),
_ => false,
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum ManifestError {
TooLarge {
max_bytes: usize,
},
MalformedJson,
NotAnObject,
NonCanonicalEncoding,
FloatValue,
MissingFormat,
UnsupportedFormat {
format: u16,
supported: u16,
},
MalformedGraph,
GraphOutOfBounds {
max_nodes: usize,
max_transitions: usize,
},
InvalidJobName,
DigestMismatch,
}
impl fmt::Display for ManifestError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::TooLarge { max_bytes } => {
write!(formatter, "definition manifest exceeds {max_bytes} bytes")
}
Self::MalformedJson => formatter.write_str("definition manifest is not valid JSON"),
Self::NotAnObject => formatter.write_str("definition manifest is not a JSON object"),
Self::NonCanonicalEncoding => {
formatter.write_str("definition manifest is not canonically encoded")
}
Self::FloatValue => {
formatter.write_str("definition manifest contains a floating-point value")
}
Self::MissingFormat => {
formatter.write_str("definition manifest has no usable format member")
}
Self::UnsupportedFormat { format, supported } => write!(
formatter,
"definition manifest format {format} is newer than the supported format {supported}"
),
Self::MalformedGraph => {
formatter.write_str("flow manifest has no readable node and transition members")
}
Self::GraphOutOfBounds {
max_nodes,
max_transitions,
} => write!(
formatter,
"flow manifest exceeds {max_nodes} nodes or {max_transitions} transitions"
),
Self::InvalidJobName => {
formatter.write_str("definition manifest binds an invalid job name")
}
Self::DigestMismatch => {
formatter.write_str("definition manifest does not match its fingerprint")
}
}
}
}
impl Error for ManifestError {}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum DefinitionTokenKind {
Revision,
Component,
Upgrade,
Classifier,
Node,
Decider,
}
#[derive(Clone, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum DefinitionError {
ZeroStartLimit,
EmptyToken {
kind: DefinitionTokenKind,
},
TokenTooLong {
kind: DefinitionTokenKind,
max_bytes: usize,
},
SurroundingWhitespace {
kind: DefinitionTokenKind,
},
ControlCharacter {
kind: DefinitionTokenKind,
},
ManifestEncoding,
ManifestTooLarge {
max_bytes: usize,
},
UpgradeSelfEdge,
EmptyStepMapping,
DuplicateSourceStep,
DuplicateTargetStep,
DeliveryModeMismatch,
CompatibilityLowering,
}
impl fmt::Display for DefinitionError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::ZeroStartLimit => formatter.write_str("start limit must be nonzero"),
Self::EmptyToken { kind } => write!(formatter, "{kind:?} token must not be empty"),
Self::TokenTooLong { kind, max_bytes } => {
write!(formatter, "{kind:?} token exceeds {max_bytes} bytes")
}
Self::SurroundingWhitespace { kind } => {
write!(formatter, "{kind:?} token has surrounding whitespace")
}
Self::ControlCharacter { kind } => {
write!(formatter, "{kind:?} token contains a control character")
}
Self::ManifestEncoding => formatter.write_str("definition manifest encoding failed"),
Self::ManifestTooLarge { max_bytes } => {
write!(formatter, "definition manifest exceeds {max_bytes} bytes")
}
Self::UpgradeSelfEdge => formatter.write_str("definition upgrade is a self-edge"),
Self::EmptyStepMapping => formatter.write_str("definition upgrade has no step mapping"),
Self::DuplicateSourceStep => {
formatter.write_str("definition upgrade repeats a source step")
}
Self::DuplicateTargetStep => {
formatter.write_str("definition upgrade reuses a target step")
}
Self::DeliveryModeMismatch => formatter
.write_str("fault runtime and restart contract declare different delivery modes"),
Self::CompatibilityLowering => {
formatter.write_str("one-step compatibility lowering produced an invalid plan")
}
}
}
}
impl Error for DefinitionError {}