ic_backup/model/execution_workflow/
mod.rs

1//! Original stage allocations and exact learned-plan bindings; no second spending ledger.
2
3use crate::model::evidence_digest::EvidenceDigest;
4use crate::model::{
5    artifacts::ArtifactChecksumRecord,
6    bounded_sequence,
7    effect_graph::MAX_EFFECT_DEPENDENCIES,
8    operation_plan::{OperationPlanError, OperationPlanRecord, PlannedOperationRecord},
9};
10use serde::{Deserialize, Deserializer, Serialize};
11use thiserror::Error;
12
13/// Maximum raw and canonical workflow declaration bytes.
14pub const MAX_EXECUTION_WORKFLOW_BYTES: u64 = 1024 * 1024;
15/// Maximum raw and canonical learned-stage binding bytes.
16pub const MAX_EXECUTION_STAGE_BYTES: u64 = 512 * 1024;
17
18/// Strict v1 original stage catalog, admitted by the existing plan structural owner.
19///
20/// Each allocation operation denotes one stage: its request hash commits the
21/// integration-owned input/purpose contract, its target is the exact stage target,
22/// and its budget is the original ceiling for that stage's later exact child plan.
23/// The allocation is never an executable operation plan or an attempt-journal
24/// authority. Only bound child plans own attempt journals. Unassigned aggregate
25/// headroom cannot be transferred to a stage. This distinct record does not change
26/// the meaning or schema of ordinary v1 operation plans.
27#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
28#[serde(try_from = "WorkflowFields")]
29pub struct ExecutionWorkflowRecord {
30    version: u16,
31    allocation: OperationPlanRecord,
32}
33#[derive(Deserialize)]
34#[serde(deny_unknown_fields)]
35struct WorkflowFields {
36    version: u16,
37    allocation: OperationPlanRecord,
38}
39impl TryFrom<WorkflowFields> for ExecutionWorkflowRecord {
40    type Error = ExecutionWorkflowError;
41    fn try_from(fields: WorkflowFields) -> Result<Self, Self::Error> {
42        if fields.version != 1 {
43            return Err(ExecutionWorkflowError::UnsupportedVersion(fields.version));
44        }
45        Ok(Self::new(fields.allocation))
46    }
47}
48impl ExecutionWorkflowRecord {
49    pub(crate) const fn allocation(&self) -> &OperationPlanRecord {
50        &self.allocation
51    }
52    /// Retain a canonical allocation without creating journals or dispatch authority.
53    #[must_use]
54    pub const fn new(allocation: OperationPlanRecord) -> Self {
55        Self {
56            version: 1,
57            allocation,
58        }
59    }
60    /// Read one original stage's purpose commitment, target and attempt ceilings.
61    /// # Errors
62    /// Rejects a stage not present in the original catalog.
63    pub fn stage(&self, sequence: u64) -> Result<&PlannedOperationRecord, OperationPlanError> {
64        self.allocation.operation(sequence)
65    }
66    /// Read the canonical original stage catalog; rows are declarations, not journals.
67    #[must_use]
68    pub fn stages(&self) -> &[PlannedOperationRecord] {
69        self.allocation.operations()
70    }
71    /// Bind the full original allocation under a distinct NUL-terminated v1 domain.
72    #[must_use]
73    pub fn digest(&self) -> ArtifactChecksumRecord {
74        let mut bytes = EvidenceDigest::new(b"ic-backup/execution-workflow/v1\0");
75        bytes.append(self.allocation.digest().hash().as_bytes());
76        bytes.finish()
77    }
78}
79
80/// Exact preceding stage binding, chronological settlement and learned-input commitment.
81///
82/// The evidence digest is integration-owned. It proves neither authenticity nor
83/// that learned snapshot IDs/dimensions were derived from those settled attempts.
84#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
85#[serde(deny_unknown_fields)]
86pub struct ExecutionStagePredecessorRecord {
87    stage_sequence: u64,
88    binding: ArtifactChecksumRecord,
89    settlement: ArtifactChecksumRecord,
90    learned_evidence: ArtifactChecksumRecord,
91}
92impl ExecutionStagePredecessorRecord {
93    /// Declare exact predecessor identities without inferring an effect outcome.
94    #[must_use]
95    pub const fn new(
96        stage_sequence: u64,
97        binding: ArtifactChecksumRecord,
98        settlement: ArtifactChecksumRecord,
99        learned_evidence: ArtifactChecksumRecord,
100    ) -> Self {
101        Self {
102            stage_sequence,
103            binding,
104            settlement,
105            learned_evidence,
106        }
107    }
108    /// Read the original predecessor stage identity.
109    #[must_use]
110    pub const fn stage_sequence(&self) -> u64 {
111        self.stage_sequence
112    }
113    /// Read the exact immutable predecessor binding identity.
114    #[must_use]
115    pub const fn binding(&self) -> &ArtifactChecksumRecord {
116        &self.binding
117    }
118    /// Read the exact retained chronological settlement identity.
119    #[must_use]
120    pub const fn settlement(&self) -> &ArtifactChecksumRecord {
121        &self.settlement
122    }
123    /// Read the opaque learned-input commitment, without qualifying its meaning.
124    #[must_use]
125    pub const fn learned_evidence(&self) -> &ArtifactChecksumRecord {
126        &self.learned_evidence
127    }
128}
129
130/// Strict v1 immutable association of one allocated stage with one exact ordinary plan.
131///
132/// The canonical child plan remains the sole request/authority owner. This record
133/// copies no counters, receipts, completion flags or payloads. Local persistence
134/// additionally requires every declared predecessor's complete Applied journal
135/// settlement; that still grants no authentic capture, transfer or dispatch permit.
136#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
137#[serde(try_from = "StageFields")]
138pub struct ExecutionStageBindingRecord {
139    version: u16,
140    workflow: ArtifactChecksumRecord,
141    stage_sequence: u64,
142    plan: ArtifactChecksumRecord,
143    predecessors: Vec<ExecutionStagePredecessorRecord>,
144}
145#[derive(Deserialize)]
146#[serde(deny_unknown_fields)]
147struct StageFields {
148    version: u16,
149    workflow: ArtifactChecksumRecord,
150    stage_sequence: u64,
151    plan: ArtifactChecksumRecord,
152    #[serde(deserialize_with = "bounded_predecessors")]
153    predecessors: Vec<ExecutionStagePredecessorRecord>,
154}
155impl TryFrom<StageFields> for ExecutionStageBindingRecord {
156    type Error = ExecutionWorkflowError;
157    fn try_from(mut fields: StageFields) -> Result<Self, Self::Error> {
158        if fields.version != 1 {
159            return Err(ExecutionWorkflowError::UnsupportedVersion(fields.version));
160        }
161        canonical_predecessors(&mut fields.predecessors)?;
162        Ok(Self {
163            version: 1,
164            workflow: fields.workflow,
165            stage_sequence: fields.stage_sequence,
166            plan: fields.plan,
167            predecessors: fields.predecessors,
168        })
169    }
170}
171impl ExecutionStageBindingRecord {
172    /// Bind exact learned requests within one original target and allocation.
173    ///
174    /// Child aggregate ceilings must equal the stage's original ceilings; existing
175    /// plan admission bounds assigned operations below those ceilings. Zero-budget
176    /// child operations reject, so the original 65,536 workflow attempt ceiling
177    /// also bounds total bound child operations. No allowance is borrowed/reset.
178    /// # Errors
179    /// Rejects changed context/inventory/target/limits, unknown stages and incomplete dependencies.
180    pub fn new(
181        workflow: &ExecutionWorkflowRecord,
182        stage_sequence: u64,
183        plan: &OperationPlanRecord,
184        mut predecessors: Vec<ExecutionStagePredecessorRecord>,
185    ) -> Result<Self, ExecutionWorkflowError> {
186        let stage = workflow
187            .stage(stage_sequence)
188            .map_err(|_| ExecutionWorkflowError::UnknownStage(stage_sequence))?;
189        if plan.context() != workflow.allocation.context()
190            || plan.inventory() != workflow.allocation.inventory()
191            || plan.selected_targets() != [stage.target()]
192        {
193            return Err(ExecutionWorkflowError::ChildIdentityMismatch);
194        }
195        if plan.budget().mutations() != stage.budget().mutations()
196            || plan.budget().observations() != stage.budget().observations()
197        {
198            return Err(ExecutionWorkflowError::ChildBudgetMismatch);
199        }
200        if plan.operations().iter().any(|operation| {
201            operation.budget().mutations() == 0 && operation.budget().observations() == 0
202        }) {
203            return Err(ExecutionWorkflowError::UnallocatedOperation);
204        }
205        canonical_predecessors(&mut predecessors)?;
206        let dependencies = workflow
207            .allocation
208            .graph()
209            .node(stage_sequence)
210            .map_err(|_| ExecutionWorkflowError::PredecessorMismatch)?
211            .depends_on();
212        if !predecessors
213            .iter()
214            .map(ExecutionStagePredecessorRecord::stage_sequence)
215            .eq(dependencies.iter().copied())
216        {
217            return Err(ExecutionWorkflowError::PredecessorMismatch);
218        }
219        Ok(Self {
220            version: 1,
221            workflow: workflow.digest(),
222            stage_sequence,
223            plan: plan.digest(),
224            predecessors,
225        })
226    }
227    /// Re-admit decoded bindings against full original workflow and child plan.
228    /// # Errors
229    /// Rejects original identity, plan or allocation/dependency changes.
230    pub fn validate(
231        &self,
232        workflow: &ExecutionWorkflowRecord,
233        plan: &OperationPlanRecord,
234    ) -> Result<(), ExecutionWorkflowError> {
235        if self.workflow != workflow.digest() || self.plan != plan.digest() {
236            return Err(ExecutionWorkflowError::BindingMismatch);
237        }
238        Self::new(
239            workflow,
240            self.stage_sequence,
241            plan,
242            self.predecessors.clone(),
243        )?;
244        Ok(())
245    }
246    /// Read full original workflow identity.
247    #[must_use]
248    pub const fn workflow(&self) -> &ArtifactChecksumRecord {
249        &self.workflow
250    }
251    /// Read the fixed original stage identity.
252    #[must_use]
253    pub const fn stage_sequence(&self) -> u64 {
254        self.stage_sequence
255    }
256    /// Read the exact canonical child plan identity.
257    #[must_use]
258    pub const fn plan(&self) -> &ArtifactChecksumRecord {
259        &self.plan
260    }
261    /// Read canonical exact predecessors and learned-input commitments.
262    #[must_use]
263    pub fn predecessors(&self) -> &[ExecutionStagePredecessorRecord] {
264        &self.predecessors
265    }
266    /// Hash the v1 NUL domain, workflow/plan ASCII hashes, big-endian u64 stage/count,
267    /// then canonical predecessor u64 identities and binding/settlement/evidence hashes.
268    #[must_use]
269    pub fn digest(&self) -> ArtifactChecksumRecord {
270        let mut bytes = EvidenceDigest::new(b"ic-backup/execution-stage/v1\0");
271        bytes.append(self.workflow.hash().as_bytes());
272        bytes.append(&self.stage_sequence.to_be_bytes());
273        bytes.append(self.plan.hash().as_bytes());
274        bytes.append(&(self.predecessors.len() as u64).to_be_bytes());
275        for row in &self.predecessors {
276            bytes.append(&row.stage_sequence.to_be_bytes());
277            for digest in [&row.binding, &row.settlement, &row.learned_evidence] {
278                bytes.append(digest.hash().as_bytes());
279            }
280        }
281        bytes.finish()
282    }
283}
284fn canonical_predecessors(
285    rows: &mut [ExecutionStagePredecessorRecord],
286) -> Result<(), ExecutionWorkflowError> {
287    if rows.len() > MAX_EFFECT_DEPENDENCIES {
288        return Err(ExecutionWorkflowError::PredecessorMismatch);
289    }
290    rows.sort_by_key(ExecutionStagePredecessorRecord::stage_sequence);
291    if rows
292        .windows(2)
293        .any(|pair| pair[0].stage_sequence == pair[1].stage_sequence)
294    {
295        return Err(ExecutionWorkflowError::PredecessorMismatch);
296    }
297    Ok(())
298}
299fn bounded_predecessors<'de, D: Deserializer<'de>>(
300    deserializer: D,
301) -> Result<Vec<ExecutionStagePredecessorRecord>, D::Error> {
302    bounded_sequence::deserialize::<_, _, MAX_EFFECT_DEPENDENCIES>(
303        deserializer,
304        "at most 1,024 exact stage predecessors",
305        ExecutionWorkflowError::PredecessorMismatch,
306    )
307}
308/// Typed rejection without changing original records or spending.
309#[derive(Debug, Error, Eq, PartialEq)]
310pub enum ExecutionWorkflowError {
311    /// Only the distinct maintained v1 generation is admitted.
312    #[error("unsupported execution workflow/stage version {0}")]
313    UnsupportedVersion(u16),
314    /// Full context, inventory or exact single allocated target changed.
315    #[error("child plan differs from original stage identity")]
316    ChildIdentityMismatch,
317    /// Child aggregate ceilings differ from original assigned stage limits.
318    #[error("child plan differs from original stage budget")]
319    ChildBudgetMismatch,
320    /// A child operation has no assigned allowance.
321    #[error("child plan contains an unallocated operation")]
322    UnallocatedOperation,
323    /// Predecessors are excessive, duplicated, missing or outside the original graph.
324    #[error("stage predecessors differ from original dependencies")]
325    PredecessorMismatch,
326    /// Retained full workflow or child plan digest differs.
327    #[error("execution stage binding mismatch")]
328    BindingMismatch,
329    /// The original catalog has no stage with this identity.
330    #[error("unknown execution stage {0}")]
331    UnknownStage(u64),
332}
333
334#[cfg(test)]
335pub(crate) mod tests;
ic_backup/model/execution_workflow/mod.rs

ic_backup/model/execution_workflow/
mod.rs
