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        validate_child(workflow, stage_sequence, plan)?;
187        canonical_predecessors(&mut predecessors)?;
188        validate_predecessor_set(workflow, stage_sequence, &predecessors)?;
189        Ok(Self {
190            version: 1,
191            workflow: workflow.digest(),
192            stage_sequence,
193            plan: plan.digest(),
194            predecessors,
195        })
196    }
197    /// Re-admit decoded bindings against full original workflow and child plan.
198    /// # Errors
199    /// Rejects original identity, plan or allocation/dependency changes.
200    pub fn validate(
201        &self,
202        workflow: &ExecutionWorkflowRecord,
203        plan: &OperationPlanRecord,
204    ) -> Result<(), ExecutionWorkflowError> {
205        if self.workflow != workflow.digest() || self.plan != plan.digest() {
206            return Err(ExecutionWorkflowError::BindingMismatch);
207        }
208        // Construction and deserialization already admit canonical predecessors.
209        // Recheck their relation to the original allocation without rebuilding a record.
210        validate_child(workflow, self.stage_sequence, plan)?;
211        validate_predecessor_set(workflow, self.stage_sequence, &self.predecessors)?;
212        Ok(())
213    }
214
215    // Both transfer plans bind one exact learned metadata/allocation predecessor.
216    // Callers first validate this binding against its full workflow and child plan.
217    pub(crate) fn matches_learned_predecessor(
218        &self,
219        plan: &OperationPlanRecord,
220        request: &str,
221        learned: &ArtifactChecksumRecord,
222        predecessors: &[ExecutionStagePredecessorRecord],
223    ) -> bool {
224        matches!(plan.operations(), [operation] if operation.request() == request)
225            && predecessors.iter().any(|row| {
226                row.stage_sequence() == self.stage_sequence
227                    && row.binding() == &self.digest()
228                    && row.learned_evidence() == learned
229            })
230    }
231
232    pub(crate) fn has_learned_evidence(
233        &self,
234        sequence: u64,
235        learned: &ArtifactChecksumRecord,
236    ) -> bool {
237        self.stage_sequence == sequence
238            && self
239                .predecessors
240                .iter()
241                .any(|row| row.learned_evidence() == learned)
242    }
243    /// Read full original workflow identity.
244    #[must_use]
245    pub const fn workflow(&self) -> &ArtifactChecksumRecord {
246        &self.workflow
247    }
248    /// Read the fixed original stage identity.
249    #[must_use]
250    pub const fn stage_sequence(&self) -> u64 {
251        self.stage_sequence
252    }
253    /// Read the exact canonical child plan identity.
254    #[must_use]
255    pub const fn plan(&self) -> &ArtifactChecksumRecord {
256        &self.plan
257    }
258    /// Read canonical exact predecessors and learned-input commitments.
259    #[must_use]
260    pub fn predecessors(&self) -> &[ExecutionStagePredecessorRecord] {
261        &self.predecessors
262    }
263    /// Hash the v1 NUL domain, workflow/plan ASCII hashes, big-endian u64 stage/count,
264    /// then canonical predecessor u64 identities and binding/settlement/evidence hashes.
265    #[must_use]
266    pub fn digest(&self) -> ArtifactChecksumRecord {
267        let mut bytes = EvidenceDigest::new(b"ic-backup/execution-stage/v1\0");
268        bytes.append(self.workflow.hash().as_bytes());
269        bytes.append(&self.stage_sequence.to_be_bytes());
270        bytes.append(self.plan.hash().as_bytes());
271        bytes.append(&(self.predecessors.len() as u64).to_be_bytes());
272        for row in &self.predecessors {
273            bytes.append(&row.stage_sequence.to_be_bytes());
274            for digest in [&row.binding, &row.settlement, &row.learned_evidence] {
275                bytes.append(digest.hash().as_bytes());
276            }
277        }
278        bytes.finish()
279    }
280}
281fn validate_child(
282    workflow: &ExecutionWorkflowRecord,
283    stage_sequence: u64,
284    plan: &OperationPlanRecord,
285) -> Result<(), ExecutionWorkflowError> {
286    let stage = workflow
287        .stage(stage_sequence)
288        .map_err(|_| ExecutionWorkflowError::UnknownStage(stage_sequence))?;
289    if plan.context() != workflow.allocation.context()
290        || plan.inventory() != workflow.allocation.inventory()
291        || plan.selected_targets() != [stage.target()]
292    {
293        return Err(ExecutionWorkflowError::ChildIdentityMismatch);
294    }
295    if plan.budget().mutations() != stage.budget().mutations()
296        || plan.budget().observations() != stage.budget().observations()
297    {
298        return Err(ExecutionWorkflowError::ChildBudgetMismatch);
299    }
300    if plan.operations().iter().any(|operation| {
301        operation.budget().mutations() == 0 && operation.budget().observations() == 0
302    }) {
303        return Err(ExecutionWorkflowError::UnallocatedOperation);
304    }
305    Ok(())
306}
307
308fn validate_predecessor_set(
309    workflow: &ExecutionWorkflowRecord,
310    stage_sequence: u64,
311    predecessors: &[ExecutionStagePredecessorRecord],
312) -> Result<(), ExecutionWorkflowError> {
313    let dependencies = workflow
314        .allocation
315        .graph()
316        .node(stage_sequence)
317        .map_err(|_| ExecutionWorkflowError::PredecessorMismatch)?
318        .depends_on();
319    if !predecessors
320        .iter()
321        .map(ExecutionStagePredecessorRecord::stage_sequence)
322        .eq(dependencies.iter().copied())
323    {
324        return Err(ExecutionWorkflowError::PredecessorMismatch);
325    }
326    Ok(())
327}
328
329fn canonical_predecessors(
330    rows: &mut [ExecutionStagePredecessorRecord],
331) -> Result<(), ExecutionWorkflowError> {
332    if rows.len() > MAX_EFFECT_DEPENDENCIES {
333        return Err(ExecutionWorkflowError::PredecessorMismatch);
334    }
335    rows.sort_by_key(ExecutionStagePredecessorRecord::stage_sequence);
336    if rows
337        .windows(2)
338        .any(|pair| pair[0].stage_sequence == pair[1].stage_sequence)
339    {
340        return Err(ExecutionWorkflowError::PredecessorMismatch);
341    }
342    Ok(())
343}
344fn bounded_predecessors<'de, D: Deserializer<'de>>(
345    deserializer: D,
346) -> Result<Vec<ExecutionStagePredecessorRecord>, D::Error> {
347    bounded_sequence::deserialize::<_, _, MAX_EFFECT_DEPENDENCIES>(
348        deserializer,
349        "at most 1,024 exact stage predecessors",
350        ExecutionWorkflowError::PredecessorMismatch,
351    )
352}
353/// Typed rejection without changing original records or spending.
354#[derive(Debug, Error, Eq, PartialEq)]
355pub enum ExecutionWorkflowError {
356    /// Only the distinct maintained v1 generation is admitted.
357    #[error("unsupported execution workflow/stage version {0}")]
358    UnsupportedVersion(u16),
359    /// Full context, inventory or exact single allocated target changed.
360    #[error("child plan differs from original stage identity")]
361    ChildIdentityMismatch,
362    /// Child aggregate ceilings differ from original assigned stage limits.
363    #[error("child plan differs from original stage budget")]
364    ChildBudgetMismatch,
365    /// A child operation has no assigned allowance.
366    #[error("child plan contains an unallocated operation")]
367    UnallocatedOperation,
368    /// Predecessors are excessive, duplicated, missing or outside the original graph.
369    #[error("stage predecessors differ from original dependencies")]
370    PredecessorMismatch,
371    /// Retained full workflow or child plan digest differs.
372    #[error("execution stage binding mismatch")]
373    BindingMismatch,
374    /// The original catalog has no stage with this identity.
375    #[error("unknown execution stage {0}")]
376    UnknownStage(u64),
377}
378
379#[cfg(test)]
380pub(crate) mod tests;
ic_backup/model/execution_workflow/mod.rs

ic_backup/model/execution_workflow/
mod.rs
