Skip to main content

oxide_batch_core/
definition.rs

1//! Restart-relevant job-definition identity.
2
3use std::collections::{BTreeMap, BTreeSet};
4use std::error::Error;
5use std::fmt;
6
7use serde_json::json;
8use sha2::{Digest, Sha256};
9
10use crate::{ChunkSize, JobName, StateSchemaId, StateSchemaVersion, StepName};
11
12const MAX_TOKEN_BYTES: usize = 128;
13/// The maximum number of nodes one plan may contain.
14///
15/// The bound is a framework capability rather than durable meaning: the
16/// manifest reader enforces the ceiling of the running build, and raising it
17/// in a later release must not change a fingerprint.
18pub const MAX_NODES: usize = 1_024;
19/// The maximum number of transitions one plan may contain.
20///
21/// The bound is a framework capability, on the same terms as [`MAX_NODES`].
22pub const MAX_TRANSITIONS: usize = 4_096;
23pub(crate) const MAX_MANIFEST_BYTES: usize = 64 * 1024;
24/// The canonical manifest format for a one-step tasklet or chunk definition.
25pub const MANIFEST_FORMAT_ONE_STEP: u16 = 1;
26/// The canonical manifest format that captures a compiled flow graph.
27pub const MANIFEST_FORMAT_FLOW: u16 = 2;
28/// The canonical manifest format for bounded M4 local-scale plans.
29pub const MANIFEST_FORMAT_LOCAL_SCALE: u16 = 3;
30/// The newest canonical manifest format this runtime can interpret.
31pub(crate) const SUPPORTED_MANIFEST_FORMAT: u16 = MANIFEST_FORMAT_LOCAL_SCALE;
32const LEGACY_REVISION: &str = "__m1_repository_port_v1";
33const LEGACY_MANIFEST: &[u8] =
34    br#"{"format":1,"repository_port":"m1","revision":"__m1_repository_port_v1"}"#;
35
36/// Validates one application-owned definition token.
37///
38/// # Errors
39///
40/// Rejects empty values, values longer than 128 UTF-8 bytes, surrounding
41/// whitespace, and control characters.
42pub fn validate_token(value: &str, kind: DefinitionTokenKind) -> Result<(), DefinitionError> {
43    if value.is_empty() {
44        return Err(DefinitionError::EmptyToken { kind });
45    }
46    if value.len() > MAX_TOKEN_BYTES {
47        return Err(DefinitionError::TokenTooLong {
48            kind,
49            max_bytes: MAX_TOKEN_BYTES,
50        });
51    }
52    if value.trim() != value {
53        return Err(DefinitionError::SurroundingWhitespace { kind });
54    }
55    if value.chars().any(char::is_control) {
56        return Err(DefinitionError::ControlCharacter { kind });
57    }
58    Ok(())
59}
60
61/// Declares one validated, application-owned definition token type.
62///
63/// The generated constructor calls [`validate_token`] and returns
64/// [`DefinitionError`], both of which must be in scope where the macro is
65/// invoked.
66#[macro_export]
67macro_rules! definition_token {
68    ($name:ident, $kind:expr, $docs:literal) => {
69        #[doc = $docs]
70        #[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
71        pub struct $name(String);
72
73        impl $name {
74            /// Validates and constructs the token.
75            ///
76            /// # Errors
77            ///
78            /// Rejects empty values, values longer than 128 UTF-8 bytes,
79            /// surrounding whitespace, and control characters.
80            pub fn new(value: impl Into<String>) -> Result<Self, DefinitionError> {
81                let value = value.into();
82                validate_token(&value, $kind)?;
83                Ok(Self(value))
84            }
85
86            /// Borrows the validated token.
87            #[must_use]
88            pub fn as_str(&self) -> &str {
89                &self.0
90            }
91        }
92    };
93}
94
95definition_token!(
96    DefinitionRevision,
97    DefinitionTokenKind::Revision,
98    "An application-owned audit label for one restart-relevant definition."
99);
100definition_token!(
101    DefinitionUpgradeKey,
102    DefinitionTokenKind::Upgrade,
103    "An application-owned key for one directed definition compatibility edge."
104);
105
106/// One source-to-target durable step mapping for a compatible restart.
107#[derive(Clone, Debug, Eq, PartialEq)]
108pub struct StepDefinitionUpgrade {
109    source: StepName,
110    target: StepName,
111}
112
113impl StepDefinitionUpgrade {
114    /// Constructs one directed step mapping.
115    #[must_use]
116    pub const fn new(source: StepName, target: StepName) -> Self {
117        Self { source, target }
118    }
119
120    /// Borrows the checkpoint-producing step name.
121    #[must_use]
122    pub const fn source(&self) -> &StepName {
123        &self.source
124    }
125
126    /// Borrows the step name in the proposed definition.
127    #[must_use]
128    pub const fn target(&self) -> &StepName {
129        &self.target
130    }
131}
132
133/// One explicit, directed definition compatibility edge.
134#[derive(Clone, Debug, Eq, PartialEq)]
135pub struct DefinitionUpgrade {
136    key: DefinitionUpgradeKey,
137    from: DefinitionIdentity,
138    to: DefinitionIdentity,
139    step_mapping: BTreeMap<StepName, StepName>,
140}
141
142impl DefinitionUpgrade {
143    /// Validates a direct, non-transitive compatibility edge.
144    ///
145    /// The M2 edge preserves checkpoint and context bytes unchanged, so
146    /// applications may use it only when the mapped steps retain the same
147    /// state schemas and semantics.
148    ///
149    /// # Errors
150    ///
151    /// Rejects self-edges, empty mappings, and mappings that reuse a target.
152    pub fn new(
153        key: DefinitionUpgradeKey,
154        from: DefinitionIdentity,
155        to: DefinitionIdentity,
156        steps: impl IntoIterator<Item = StepDefinitionUpgrade>,
157    ) -> Result<Self, DefinitionError> {
158        if from.manifest_digest() == to.manifest_digest() {
159            return Err(DefinitionError::UpgradeSelfEdge);
160        }
161        let mut step_mapping = BTreeMap::new();
162        let mut targets = BTreeSet::new();
163        for step in steps {
164            if step_mapping
165                .insert(step.source().clone(), step.target().clone())
166                .is_some()
167            {
168                return Err(DefinitionError::DuplicateSourceStep);
169            }
170            if !targets.insert(step.target().clone()) {
171                return Err(DefinitionError::DuplicateTargetStep);
172            }
173        }
174        if step_mapping.is_empty() {
175            return Err(DefinitionError::EmptyStepMapping);
176        }
177        Ok(Self {
178            key,
179            from,
180            to,
181            step_mapping,
182        })
183    }
184
185    /// Borrows the application-owned upgrade key.
186    #[must_use]
187    pub const fn key(&self) -> &DefinitionUpgradeKey {
188        &self.key
189    }
190
191    /// Borrows the checkpoint-producing definition.
192    #[must_use]
193    pub const fn from(&self) -> &DefinitionIdentity {
194        &self.from
195    }
196
197    /// Borrows the proposed definition.
198    #[must_use]
199    pub const fn to(&self) -> &DefinitionIdentity {
200        &self.to
201    }
202
203    /// Borrows the durable source-to-target step mapping.
204    ///
205    /// Durable adapters replay the mapping when they resolve a compatible
206    /// restart, so the order is the validated declaration order.
207    #[must_use]
208    pub fn step_mapping(&self) -> &BTreeMap<StepName, StepName> {
209        &self.step_mapping
210    }
211}
212definition_token!(
213    ComponentRevision,
214    DefinitionTokenKind::Component,
215    "An application-owned revision token for one opaque executable component."
216);
217definition_token!(
218    ClassifierRevision,
219    DefinitionTokenKind::Classifier,
220    "An application-owned revision token for one bounded fault classifier."
221);
222
223/// Component revisions for a one-step chunk definition.
224#[derive(Clone, Debug, Eq, PartialEq)]
225pub struct ChunkComponentRevisions {
226    reader: ComponentRevision,
227    processor: ComponentRevision,
228    writer: ComponentRevision,
229    checkpoint: ComponentRevision,
230    restart: ChunkRestartContract,
231}
232
233impl ChunkComponentRevisions {
234    /// Constructs the four restart-relevant chunk component revisions.
235    #[must_use]
236    pub const fn new(
237        reader: ComponentRevision,
238        processor: ComponentRevision,
239        writer: ComponentRevision,
240        checkpoint: ComponentRevision,
241        restart: ChunkRestartContract,
242    ) -> Self {
243        Self {
244            reader,
245            processor,
246            writer,
247            checkpoint,
248            restart,
249        }
250    }
251
252    /// Returns the delivery mode declared by the restart contract.
253    #[must_use]
254    pub const fn delivery_mode(&self) -> ChunkDeliveryMode {
255        self.restart.delivery_mode
256    }
257
258    /// Returns the shutdown policy of an already-open chunk.
259    #[must_use]
260    pub const fn in_flight_policy(&self) -> InFlightPolicy {
261        self.restart.in_flight_policy
262    }
263
264    /// Borrows the revision of the component that reads items.
265    #[must_use]
266    pub const fn reader(&self) -> &ComponentRevision {
267        &self.reader
268    }
269
270    /// Borrows the revision of the component that processes items.
271    #[must_use]
272    pub const fn processor(&self) -> &ComponentRevision {
273        &self.processor
274    }
275
276    /// Borrows the revision of the component that writes items.
277    #[must_use]
278    pub const fn writer(&self) -> &ComponentRevision {
279        &self.writer
280    }
281
282    /// Borrows the revision of the component that produces checkpoints.
283    #[must_use]
284    pub const fn checkpoint(&self) -> &ComponentRevision {
285        &self.checkpoint
286    }
287
288    /// Borrows the declared checkpoint schema.
289    #[must_use]
290    pub const fn checkpoint_schema(&self) -> &StateSchemaId {
291        &self.restart.checkpoint_schema
292    }
293
294    /// Returns the declared checkpoint schema version.
295    #[must_use]
296    pub const fn checkpoint_schema_version(&self) -> StateSchemaVersion {
297        self.restart.checkpoint_schema_version
298    }
299
300    /// Borrows the declared execution-context schema.
301    #[must_use]
302    pub const fn context_schema(&self) -> &StateSchemaId {
303        &self.restart.context_schema
304    }
305
306    /// Returns the declared execution-context schema version.
307    #[must_use]
308    pub const fn context_schema_version(&self) -> StateSchemaVersion {
309        self.restart.context_schema_version
310    }
311}
312
313/// The accepted shutdown behavior for an already-open chunk.
314#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
315#[non_exhaustive]
316pub enum InFlightPolicy {
317    /// Complete and commit the open chunk, then stop at its boundary.
318    #[default]
319    FinishChunk,
320    /// Roll back the open chunk and preserve the prior checkpoint.
321    RollbackChunk,
322}
323
324/// Declared delivery boundary included in a chunk definition fingerprint.
325#[derive(Clone, Copy, Debug, Eq, PartialEq)]
326#[non_exhaustive]
327pub enum ChunkDeliveryMode {
328    /// Business writes and progress share one `PostgreSQL` transaction.
329    AtomicSameResource,
330    /// The resource may observe a duplicate after restart.
331    AtLeastOnce,
332}
333
334impl ChunkDeliveryMode {
335    /// Returns the stable name this mode is recorded under in a manifest.
336    ///
337    /// The name is durable: it is hashed into a definition fingerprint, so it
338    /// is fixed for the life of the mode rather than a display string.
339    #[must_use]
340    pub const fn manifest_name(self) -> &'static str {
341        match self {
342            Self::AtomicSameResource => "atomic_same_resource",
343            Self::AtLeastOnce => "at_least_once",
344        }
345    }
346}
347
348/// Restart-state schemas and delivery mode for a chunk definition.
349#[derive(Clone, Debug, Eq, PartialEq)]
350pub struct ChunkRestartContract {
351    checkpoint_schema: StateSchemaId,
352    checkpoint_schema_version: StateSchemaVersion,
353    context_schema: StateSchemaId,
354    context_schema_version: StateSchemaVersion,
355    delivery_mode: ChunkDeliveryMode,
356    in_flight_policy: InFlightPolicy,
357}
358
359impl ChunkRestartContract {
360    /// Constructs the restart-relevant state and delivery declaration.
361    #[must_use]
362    pub const fn new(
363        checkpoint_schema: StateSchemaId,
364        checkpoint_schema_version: StateSchemaVersion,
365        context_schema: StateSchemaId,
366        context_schema_version: StateSchemaVersion,
367        delivery_mode: ChunkDeliveryMode,
368    ) -> Self {
369        Self {
370            checkpoint_schema,
371            checkpoint_schema_version,
372            context_schema,
373            context_schema_version,
374            delivery_mode,
375            in_flight_policy: InFlightPolicy::FinishChunk,
376        }
377    }
378
379    /// Selects the restart-relevant open-chunk shutdown policy.
380    #[must_use]
381    pub const fn with_in_flight_policy(mut self, policy: InFlightPolicy) -> Self {
382        self.in_flight_policy = policy;
383        self
384    }
385}
386
387/// Canonical restart-relevant identity persisted with every execution.
388#[derive(Clone, Eq, PartialEq)]
389pub struct DefinitionIdentity {
390    job_name: Option<JobName>,
391    revision: DefinitionRevision,
392    manifest_format: u16,
393    manifest_digest: [u8; 32],
394    canonical_manifest: Box<[u8]>,
395}
396
397impl DefinitionIdentity {
398    /// Builds the identity durable rows written before manifests carried one.
399    ///
400    /// The bytes are frozen: a durable row that predates manifest identity
401    /// must keep resolving to exactly this value.
402    #[must_use]
403    pub fn legacy() -> Self {
404        Self::from_canonical(
405            None,
406            DefinitionRevision(LEGACY_REVISION.to_owned()),
407            LEGACY_MANIFEST.to_vec(),
408            MANIFEST_FORMAT_ONE_STEP,
409        )
410    }
411
412    /// Builds the canonical identity for a one-step tasklet definition.
413    ///
414    /// # Errors
415    ///
416    /// Returns [`DefinitionError::ManifestEncoding`] if the bounded canonical
417    /// manifest cannot be encoded.
418    pub fn tasklet(
419        job_name: &JobName,
420        step_name: &StepName,
421        revision: DefinitionRevision,
422        component_revision: &ComponentRevision,
423    ) -> Result<Self, DefinitionError> {
424        let manifest = json!({
425            "component": {
426                "tasklet": component_revision.as_str()
427            },
428            "delivery_mode": "best_effort",
429            "format": MANIFEST_FORMAT_ONE_STEP,
430            "job": job_name.as_str(),
431            "kind": "tasklet",
432            "restart_state": "none",
433            "step": step_name.as_str(),
434            "transaction_boundary": "tasklet_completion"
435        });
436        Self::encode(job_name.clone(), revision, &manifest)
437    }
438
439    /// Builds the canonical identity for a one-step chunk definition.
440    ///
441    /// # Errors
442    ///
443    /// Returns [`DefinitionError::ManifestEncoding`] if the bounded canonical
444    /// manifest cannot be encoded.
445    pub fn chunk(
446        job_name: &JobName,
447        step_name: &StepName,
448        chunk_size: ChunkSize,
449        revision: DefinitionRevision,
450        components: &ChunkComponentRevisions,
451    ) -> Result<Self, DefinitionError> {
452        let mut manifest = json!({
453            "chunk_size": chunk_size.get(),
454            "components": {
455                "checkpoint": components.checkpoint.as_str(),
456                "processor": components.processor.as_str(),
457                "reader": components.reader.as_str(),
458                "writer": components.writer.as_str()
459            },
460            "context": {
461                "schema": components.restart.context_schema.as_str(),
462                "version": components.restart.context_schema_version.get()
463            },
464            "checkpoint": {
465                "schema": components.restart.checkpoint_schema.as_str(),
466                "version": components.restart.checkpoint_schema_version.get()
467            },
468            "delivery_mode": components.restart.delivery_mode.manifest_name(),
469            "format": MANIFEST_FORMAT_ONE_STEP,
470            "job": job_name.as_str(),
471            "kind": "chunk",
472            "step": step_name.as_str(),
473            "transaction_boundary": "chunk"
474        });
475        if components.restart.in_flight_policy == InFlightPolicy::RollbackChunk
476            && let Some(object) = manifest.as_object_mut()
477        {
478            object.insert(
479                "in_flight_policy".to_owned(),
480                serde_json::Value::String("rollback_chunk".to_owned()),
481            );
482        }
483        Self::encode(job_name.clone(), revision, &manifest)
484    }
485
486    /// Builds the canonical identity for a compiled flow graph.
487    ///
488    /// The bytes must already be the canonical manifest the plan compiler
489    /// normalized and encoded; this constructor validates, bounds, and hashes
490    /// them. Taking bytes rather than a parsed document keeps the serializer
491    /// out of this contract and makes the encoding the caller is responsible
492    /// for explicit rather than implied.
493    ///
494    /// # Errors
495    ///
496    /// Returns [`DefinitionError::ManifestEncoding`] when the bytes exceed the
497    /// bounded encoding, are not a canonical JSON object, or do not declare a
498    /// supported flow manifest format.
499    pub fn from_flow_manifest(
500        job_name: &JobName,
501        revision: DefinitionRevision,
502        canonical: &[u8],
503    ) -> Result<Self, DefinitionError> {
504        if canonical.len() > MAX_MANIFEST_BYTES {
505            return Err(DefinitionError::ManifestTooLarge {
506                max_bytes: MAX_MANIFEST_BYTES,
507            });
508        }
509        let document: serde_json::Value =
510            serde_json::from_slice(canonical).map_err(|_| DefinitionError::ManifestEncoding)?;
511        let reencoded =
512            serde_json::to_vec(&document).map_err(|_| DefinitionError::ManifestEncoding)?;
513        if !document.is_object() || reencoded != canonical {
514            return Err(DefinitionError::ManifestEncoding);
515        }
516        let format = document
517            .get("format")
518            .and_then(serde_json::Value::as_u64)
519            .and_then(|value| u16::try_from(value).ok())
520            .filter(|value| matches!(*value, MANIFEST_FORMAT_FLOW | MANIFEST_FORMAT_LOCAL_SCALE))
521            .ok_or(DefinitionError::ManifestEncoding)?;
522
523        Ok(Self::from_canonical(
524            Some(job_name.clone()),
525            revision,
526            canonical.to_vec(),
527            format,
528        ))
529    }
530
531    /// Encodes a one-step manifest this crate composed itself.
532    fn encode(
533        job_name: JobName,
534        revision: DefinitionRevision,
535        manifest: &serde_json::Value,
536    ) -> Result<Self, DefinitionError> {
537        let canonical =
538            serde_json::to_vec(manifest).map_err(|_| DefinitionError::ManifestEncoding)?;
539        if canonical.len() > MAX_MANIFEST_BYTES {
540            return Err(DefinitionError::ManifestTooLarge {
541                max_bytes: MAX_MANIFEST_BYTES,
542            });
543        }
544        Ok(Self::from_canonical(
545            Some(job_name),
546            revision,
547            canonical,
548            MANIFEST_FORMAT_ONE_STEP,
549        ))
550    }
551
552    fn from_canonical(
553        job_name: Option<JobName>,
554        revision: DefinitionRevision,
555        canonical: Vec<u8>,
556        format: u16,
557    ) -> Self {
558        let digest: [u8; 32] = Sha256::digest(&canonical).into();
559        Self {
560            job_name,
561            revision,
562            manifest_format: format,
563            manifest_digest: digest,
564            canonical_manifest: canonical.into_boxed_slice(),
565        }
566    }
567
568    /// Borrows the application-owned definition revision.
569    #[must_use]
570    pub const fn revision(&self) -> &DefinitionRevision {
571        &self.revision
572    }
573
574    /// Borrows the job name bound into a framework-produced manifest.
575    ///
576    /// Legacy direct repository calls use an internal compatibility manifest
577    /// without a bound name.
578    #[must_use]
579    pub const fn job_name(&self) -> Option<&JobName> {
580        self.job_name.as_ref()
581    }
582
583    /// Returns the canonical manifest format version.
584    #[must_use]
585    pub const fn manifest_format(&self) -> u16 {
586        self.manifest_format
587    }
588
589    /// Returns the framework-produced SHA-256 manifest digest.
590    #[must_use]
591    pub const fn manifest_digest(&self) -> &[u8; 32] {
592        &self.manifest_digest
593    }
594
595    /// Borrows the exact canonical manifest bytes that produced the digest.
596    ///
597    /// The manifest records names, logical identifiers, revisions, schema
598    /// versions, and bounded policy values. Parameters, contexts, item values,
599    /// credentials, endpoints, and component-private state are never encoded
600    /// into it, so operators may inspect and archive these bytes.
601    #[must_use]
602    pub fn canonical_manifest(&self) -> &[u8] {
603        &self.canonical_manifest
604    }
605}
606
607/// Returns whether this runtime can interpret `format`.
608///
609/// # Errors
610///
611/// Returns [`ManifestError::UnsupportedFormat`] for a newer format and
612/// [`ManifestError::MissingFormat`] for zero.
613pub const fn check_manifest_format(format: u16) -> Result<(), ManifestError> {
614    if format == 0 {
615        return Err(ManifestError::MissingFormat);
616    }
617    if format > SUPPORTED_MANIFEST_FORMAT {
618        return Err(ManifestError::UnsupportedFormat {
619            format,
620            supported: SUPPORTED_MANIFEST_FORMAT,
621        });
622    }
623    Ok(())
624}
625
626impl fmt::Debug for DefinitionIdentity {
627    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
628        formatter
629            .debug_struct("DefinitionIdentity")
630            .field("job_name", &self.job_name)
631            .field("revision", &self.revision)
632            .field("manifest_format", &self.manifest_format)
633            .field(
634                "digest_prefix",
635                &DigestPrefix([
636                    self.manifest_digest[0],
637                    self.manifest_digest[1],
638                    self.manifest_digest[2],
639                    self.manifest_digest[3],
640                ]),
641            )
642            .field("canonical_manifest", &"<redacted>")
643            .finish()
644    }
645}
646
647struct DigestPrefix([u8; 4]);
648
649impl fmt::Debug for DigestPrefix {
650    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
651        for byte in self.0 {
652            write!(formatter, "{byte:02x}")?;
653        }
654        Ok(())
655    }
656}
657
658/// A validated, read-only view of one canonical definition manifest.
659///
660/// The reader accepts every manifest format this runtime understands, so a
661/// deployment can inspect a definition persisted by an older release. It never
662/// guesses: a newer format, a non-canonical encoding, a floating-point value,
663/// an out-of-bound graph, or a digest that does not match the supplied bytes
664/// fails closed.
665///
666/// The runnable example lives in the `oxide-batch` facade documentation,
667/// because the supported import path is `oxide_batch`.
668#[derive(Clone, Debug, Eq, PartialEq)]
669pub struct DefinitionManifest {
670    format: u16,
671    digest: [u8; 32],
672    job_name: Option<JobName>,
673    node_count: Option<usize>,
674    transition_count: Option<usize>,
675}
676
677impl DefinitionManifest {
678    /// Reads and validates canonical manifest bytes.
679    ///
680    /// # Errors
681    ///
682    /// Returns [`ManifestError`] when the bytes exceed the durable bound, are
683    /// not a canonical JSON object, omit or overflow the format member, declare
684    /// a format this runtime cannot interpret, contain a floating-point value,
685    /// or describe a graph outside the accepted bounds.
686    pub fn read(bytes: &[u8]) -> Result<Self, ManifestError> {
687        if bytes.len() > MAX_MANIFEST_BYTES {
688            return Err(ManifestError::TooLarge {
689                max_bytes: MAX_MANIFEST_BYTES,
690            });
691        }
692        let document: serde_json::Value =
693            serde_json::from_slice(bytes).map_err(|_| ManifestError::MalformedJson)?;
694        let members = document.as_object().ok_or(ManifestError::NotAnObject)?;
695        if contains_float(&document) {
696            return Err(ManifestError::FloatValue);
697        }
698        let reencoded = serde_json::to_vec(&document).map_err(|_| ManifestError::MalformedJson)?;
699        if reencoded != bytes {
700            return Err(ManifestError::NonCanonicalEncoding);
701        }
702        let format = members
703            .get("format")
704            .and_then(serde_json::Value::as_u64)
705            .and_then(|format| u16::try_from(format).ok())
706            .ok_or(ManifestError::MissingFormat)?;
707        check_manifest_format(format)?;
708        let job_name = members
709            .get("job")
710            .and_then(serde_json::Value::as_str)
711            .map(JobName::new)
712            .transpose()
713            .map_err(|_| ManifestError::InvalidJobName)?;
714        let (node_count, transition_count) =
715            if matches!(format, MANIFEST_FORMAT_FLOW | MANIFEST_FORMAT_LOCAL_SCALE) {
716                let nodes = array_len(members.get("nodes"))?;
717                let transitions = array_len(members.get("transitions"))?;
718                if nodes > MAX_NODES || transitions > MAX_TRANSITIONS {
719                    return Err(ManifestError::GraphOutOfBounds {
720                        max_nodes: MAX_NODES,
721                        max_transitions: MAX_TRANSITIONS,
722                    });
723                }
724                (Some(nodes), Some(transitions))
725            } else {
726                (None, None)
727            };
728        Ok(Self {
729            format,
730            digest: Sha256::digest(bytes).into(),
731            job_name,
732            node_count,
733            transition_count,
734        })
735    }
736
737    /// Reads canonical bytes and requires them to hash to `expected`.
738    ///
739    /// # Errors
740    ///
741    /// Returns every [`ManifestError`] [`read`](Self::read) returns, plus
742    /// [`ManifestError::DigestMismatch`] when the bytes were altered.
743    pub fn read_verified(bytes: &[u8], expected: &[u8; 32]) -> Result<Self, ManifestError> {
744        let manifest = Self::read(bytes)?;
745        if &manifest.digest != expected {
746            return Err(ManifestError::DigestMismatch);
747        }
748        Ok(manifest)
749    }
750
751    /// Returns the declared canonical manifest format.
752    #[must_use]
753    pub const fn format(&self) -> u16 {
754        self.format
755    }
756
757    /// Returns the SHA-256 digest of the exact bytes that were read.
758    #[must_use]
759    pub const fn digest(&self) -> &[u8; 32] {
760        &self.digest
761    }
762
763    /// Borrows the job name bound into the manifest, when present.
764    #[must_use]
765    pub const fn job_name(&self) -> Option<&JobName> {
766        self.job_name.as_ref()
767    }
768
769    /// Returns the compiled node count of a flow manifest.
770    #[must_use]
771    pub const fn node_count(&self) -> Option<usize> {
772        self.node_count
773    }
774
775    /// Returns the compiled transition count of a flow manifest.
776    #[must_use]
777    pub const fn transition_count(&self) -> Option<usize> {
778        self.transition_count
779    }
780}
781
782fn array_len(value: Option<&serde_json::Value>) -> Result<usize, ManifestError> {
783    value
784        .and_then(serde_json::Value::as_array)
785        .map(Vec::len)
786        .ok_or(ManifestError::MalformedGraph)
787}
788
789fn contains_float(value: &serde_json::Value) -> bool {
790    match value {
791        serde_json::Value::Number(number) => number.as_i64().is_none() && number.as_u64().is_none(),
792        serde_json::Value::Array(values) => values.iter().any(contains_float),
793        serde_json::Value::Object(members) => members.values().any(contains_float),
794        _ => false,
795    }
796}
797
798/// A canonical definition manifest that cannot be interpreted.
799#[derive(Clone, Copy, Debug, Eq, PartialEq)]
800#[non_exhaustive]
801pub enum ManifestError {
802    /// The bytes exceeded the durable manifest bound.
803    TooLarge {
804        /// Maximum accepted byte length.
805        max_bytes: usize,
806    },
807    /// The bytes are not valid JSON.
808    MalformedJson,
809    /// The document is not a JSON object.
810    NotAnObject,
811    /// Re-encoding the document did not reproduce the supplied bytes.
812    NonCanonicalEncoding,
813    /// The document contains a floating-point number.
814    FloatValue,
815    /// The format member is absent or not a `u16`.
816    MissingFormat,
817    /// The manifest is newer than this runtime understands.
818    UnsupportedFormat {
819        /// Format found in the manifest.
820        format: u16,
821        /// Newest format this runtime interprets.
822        supported: u16,
823    },
824    /// A flow manifest omitted or malformed its graph members.
825    MalformedGraph,
826    /// A flow manifest declared a graph larger than this runtime accepts.
827    GraphOutOfBounds {
828        /// Maximum accepted node count.
829        max_nodes: usize,
830        /// Maximum accepted transition count.
831        max_transitions: usize,
832    },
833    /// The bound job name is not a valid [`JobName`].
834    InvalidJobName,
835    /// The bytes do not hash to the expected fingerprint.
836    DigestMismatch,
837}
838
839impl fmt::Display for ManifestError {
840    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
841        match self {
842            Self::TooLarge { max_bytes } => {
843                write!(formatter, "definition manifest exceeds {max_bytes} bytes")
844            }
845            Self::MalformedJson => formatter.write_str("definition manifest is not valid JSON"),
846            Self::NotAnObject => formatter.write_str("definition manifest is not a JSON object"),
847            Self::NonCanonicalEncoding => {
848                formatter.write_str("definition manifest is not canonically encoded")
849            }
850            Self::FloatValue => {
851                formatter.write_str("definition manifest contains a floating-point value")
852            }
853            Self::MissingFormat => {
854                formatter.write_str("definition manifest has no usable format member")
855            }
856            Self::UnsupportedFormat { format, supported } => write!(
857                formatter,
858                "definition manifest format {format} is newer than the supported format {supported}"
859            ),
860            Self::MalformedGraph => {
861                formatter.write_str("flow manifest has no readable node and transition members")
862            }
863            Self::GraphOutOfBounds {
864                max_nodes,
865                max_transitions,
866            } => write!(
867                formatter,
868                "flow manifest exceeds {max_nodes} nodes or {max_transitions} transitions"
869            ),
870            Self::InvalidJobName => {
871                formatter.write_str("definition manifest binds an invalid job name")
872            }
873            Self::DigestMismatch => {
874                formatter.write_str("definition manifest does not match its fingerprint")
875            }
876        }
877    }
878}
879
880impl Error for ManifestError {}
881
882/// Definition token category used by validation diagnostics.
883#[derive(Clone, Copy, Debug, Eq, PartialEq)]
884#[non_exhaustive]
885pub enum DefinitionTokenKind {
886    /// Definition revision.
887    Revision,
888    /// Opaque component revision.
889    Component,
890    /// Directed compatibility edge key.
891    Upgrade,
892    /// Bounded fault-classifier revision.
893    Classifier,
894    /// Stable flow-graph node identifier.
895    Node,
896    /// Bounded decider revision.
897    Decider,
898}
899
900/// Failure to construct a bounded restart definition.
901#[derive(Clone, Debug, Eq, PartialEq)]
902#[non_exhaustive]
903pub enum DefinitionError {
904    /// A start limit of zero can never start its step.
905    ZeroStartLimit,
906    /// A required token was empty.
907    EmptyToken {
908        /// Rejected token category.
909        kind: DefinitionTokenKind,
910    },
911    /// A token exceeded its UTF-8 byte bound.
912    TokenTooLong {
913        /// Rejected token category.
914        kind: DefinitionTokenKind,
915        /// Maximum accepted byte length.
916        max_bytes: usize,
917    },
918    /// A token had leading or trailing whitespace.
919    SurroundingWhitespace {
920        /// Rejected token category.
921        kind: DefinitionTokenKind,
922    },
923    /// A token contained a control character.
924    ControlCharacter {
925        /// Rejected token category.
926        kind: DefinitionTokenKind,
927    },
928    /// The canonical manifest could not be encoded.
929    ManifestEncoding,
930    /// The canonical manifest exceeded its durable bound.
931    ManifestTooLarge {
932        /// Maximum accepted byte length.
933        max_bytes: usize,
934    },
935    /// A directed edge pointed from a definition to itself.
936    UpgradeSelfEdge,
937    /// A directed edge omitted its durable step mapping.
938    EmptyStepMapping,
939    /// A source step appeared more than once.
940    DuplicateSourceStep,
941    /// Two source steps mapped to the same target step.
942    DuplicateTargetStep,
943    /// A step's fault runtime declared a different delivery mode than its
944    /// restart contract.
945    DeliveryModeMismatch,
946    /// A one-step wrapper could not be lowered into its compatibility plan.
947    ///
948    /// The framework derives the compatibility graph from values it has
949    /// already validated, so this variant reports a framework invariant rather
950    /// than an application mistake.
951    CompatibilityLowering,
952}
953
954impl fmt::Display for DefinitionError {
955    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
956        match self {
957            Self::ZeroStartLimit => formatter.write_str("start limit must be nonzero"),
958            Self::EmptyToken { kind } => write!(formatter, "{kind:?} token must not be empty"),
959            Self::TokenTooLong { kind, max_bytes } => {
960                write!(formatter, "{kind:?} token exceeds {max_bytes} bytes")
961            }
962            Self::SurroundingWhitespace { kind } => {
963                write!(formatter, "{kind:?} token has surrounding whitespace")
964            }
965            Self::ControlCharacter { kind } => {
966                write!(formatter, "{kind:?} token contains a control character")
967            }
968            Self::ManifestEncoding => formatter.write_str("definition manifest encoding failed"),
969            Self::ManifestTooLarge { max_bytes } => {
970                write!(formatter, "definition manifest exceeds {max_bytes} bytes")
971            }
972            Self::UpgradeSelfEdge => formatter.write_str("definition upgrade is a self-edge"),
973            Self::EmptyStepMapping => formatter.write_str("definition upgrade has no step mapping"),
974            Self::DuplicateSourceStep => {
975                formatter.write_str("definition upgrade repeats a source step")
976            }
977            Self::DuplicateTargetStep => {
978                formatter.write_str("definition upgrade reuses a target step")
979            }
980            Self::DeliveryModeMismatch => formatter
981                .write_str("fault runtime and restart contract declare different delivery modes"),
982            Self::CompatibilityLowering => {
983                formatter.write_str("one-step compatibility lowering produced an invalid plan")
984            }
985        }
986    }
987}
988
989impl Error for DefinitionError {}