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    /// Read original stage purposes, context and ceilings without deriving dispatch authority.
50    /// Child plans must still pass exact stage binding and complete journal admission.
51    #[must_use]
52    pub const fn allocation(&self) -> &OperationPlanRecord {
53        &self.allocation
54    }
55    /// Retain a canonical allocation without creating journals or dispatch authority.
56    #[must_use]
57    pub const fn new(allocation: OperationPlanRecord) -> Self {
58        Self {
59            version: 1,
60            allocation,
61        }
62    }
63    /// Read one original stage's purpose commitment, target and attempt ceilings.
64    /// # Errors
65    /// Rejects a stage not present in the original catalog.
66    pub fn stage(&self, sequence: u64) -> Result<&PlannedOperationRecord, OperationPlanError> {
67        self.allocation.operation(sequence)
68    }
69    /// Read the canonical original stage catalog; rows are declarations, not journals.
70    #[must_use]
71    pub fn stages(&self) -> &[PlannedOperationRecord] {
72        self.allocation.operations()
73    }
74    /// Bind the full original allocation under a distinct NUL-terminated v1 domain.
75    #[must_use]
76    pub fn digest(&self) -> ArtifactChecksumRecord {
77        let mut bytes = EvidenceDigest::new(b"ic-backup/execution-workflow/v1\0");
78        bytes.checksum(&self.allocation.digest());
79        bytes.finish()
80    }
81}
82
83/// Exact preceding stage binding, chronological settlement and learned-input commitment.
84///
85/// The evidence digest is integration-owned. It proves neither authenticity nor
86/// that learned snapshot IDs/dimensions were derived from those settled attempts.
87#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
88#[serde(deny_unknown_fields)]
89pub struct ExecutionStagePredecessorRecord {
90    stage_sequence: u64,
91    binding: ArtifactChecksumRecord,
92    settlement: ArtifactChecksumRecord,
93    learned_evidence: ArtifactChecksumRecord,
94}
95impl ExecutionStagePredecessorRecord {
96    /// Declare exact predecessor identities without inferring an effect outcome.
97    #[must_use]
98    pub const fn new(
99        stage_sequence: u64,
100        binding: ArtifactChecksumRecord,
101        settlement: ArtifactChecksumRecord,
102        learned_evidence: ArtifactChecksumRecord,
103    ) -> Self {
104        Self {
105            stage_sequence,
106            binding,
107            settlement,
108            learned_evidence,
109        }
110    }
111    /// Read the original predecessor stage identity.
112    #[must_use]
113    pub const fn stage_sequence(&self) -> u64 {
114        self.stage_sequence
115    }
116    /// Read the exact immutable predecessor binding identity.
117    #[must_use]
118    pub const fn binding(&self) -> &ArtifactChecksumRecord {
119        &self.binding
120    }
121    /// Read the exact retained chronological settlement identity.
122    #[must_use]
123    pub const fn settlement(&self) -> &ArtifactChecksumRecord {
124        &self.settlement
125    }
126    /// Read the opaque learned-input commitment, without qualifying its meaning.
127    #[must_use]
128    pub const fn learned_evidence(&self) -> &ArtifactChecksumRecord {
129        &self.learned_evidence
130    }
131}
132
133/// Strict v1 immutable association of one allocated stage with one exact ordinary plan.
134///
135/// The canonical child plan remains the sole request/authority owner. This record
136/// copies no counters, receipts, completion flags or payloads. Local persistence
137/// additionally requires every declared predecessor's complete Applied journal
138/// settlement; that still grants no authentic capture, transfer or dispatch permit.
139#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
140#[serde(try_from = "StageFields")]
141pub struct ExecutionStageBindingRecord {
142    version: u16,
143    workflow: ArtifactChecksumRecord,
144    stage_sequence: u64,
145    plan: ArtifactChecksumRecord,
146    predecessors: Vec<ExecutionStagePredecessorRecord>,
147}
148#[derive(Deserialize)]
149#[serde(deny_unknown_fields)]
150struct StageFields {
151    version: u16,
152    workflow: ArtifactChecksumRecord,
153    stage_sequence: u64,
154    plan: ArtifactChecksumRecord,
155    #[serde(deserialize_with = "bounded_predecessors")]
156    predecessors: Vec<ExecutionStagePredecessorRecord>,
157}
158impl TryFrom<StageFields> for ExecutionStageBindingRecord {
159    type Error = ExecutionWorkflowError;
160    fn try_from(mut fields: StageFields) -> Result<Self, Self::Error> {
161        if fields.version != 1 {
162            return Err(ExecutionWorkflowError::UnsupportedVersion(fields.version));
163        }
164        canonical_predecessors(&mut fields.predecessors)?;
165        Ok(Self {
166            version: 1,
167            workflow: fields.workflow,
168            stage_sequence: fields.stage_sequence,
169            plan: fields.plan,
170            predecessors: fields.predecessors,
171        })
172    }
173}
174impl ExecutionStageBindingRecord {
175    /// Bind exact learned requests within one original target and allocation.
176    ///
177    /// Child aggregate ceilings must equal the stage's original ceilings; existing
178    /// plan admission bounds assigned operations below those ceilings. Zero-budget
179    /// child operations reject, so the original 65,536 workflow attempt ceiling
180    /// also bounds total bound child operations. No allowance is borrowed/reset.
181    /// # Errors
182    /// Rejects changed context/inventory/target/limits, unknown stages and incomplete dependencies.
183    pub fn new(
184        workflow: &ExecutionWorkflowRecord,
185        stage_sequence: u64,
186        plan: &OperationPlanRecord,
187        mut predecessors: Vec<ExecutionStagePredecessorRecord>,
188    ) -> Result<Self, ExecutionWorkflowError> {
189        validate_child(workflow, stage_sequence, plan)?;
190        canonical_predecessors(&mut predecessors)?;
191        validate_predecessor_set(workflow, stage_sequence, &predecessors)?;
192        Ok(Self {
193            version: 1,
194            workflow: workflow.digest(),
195            stage_sequence,
196            plan: plan.digest(),
197            predecessors,
198        })
199    }
200    /// Re-admit decoded bindings against full original workflow and child plan.
201    /// # Errors
202    /// Rejects original identity, plan or allocation/dependency changes.
203    pub fn validate(
204        &self,
205        workflow: &ExecutionWorkflowRecord,
206        plan: &OperationPlanRecord,
207    ) -> Result<(), ExecutionWorkflowError> {
208        if self.workflow != workflow.digest() || self.plan != plan.digest() {
209            return Err(ExecutionWorkflowError::BindingMismatch);
210        }
211        // Construction and deserialization already admit canonical predecessors.
212        // Recheck their relation to the original allocation without rebuilding a record.
213        validate_child(workflow, self.stage_sequence, plan)?;
214        validate_predecessor_set(workflow, self.stage_sequence, &self.predecessors)?;
215        Ok(())
216    }
217
218    // Both transfer plans bind one exact learned metadata/allocation predecessor.
219    // Callers first validate this binding against its full workflow and child plan.
220    pub(crate) fn matches_learned_predecessor(
221        &self,
222        plan: &OperationPlanRecord,
223        request: &str,
224        learned: &ArtifactChecksumRecord,
225        predecessors: &[ExecutionStagePredecessorRecord],
226    ) -> bool {
227        matches!(plan.operations(), [operation] if operation.request() == request)
228            && predecessors.iter().any(|row| {
229                row.stage_sequence() == self.stage_sequence
230                    && row.binding() == &self.digest()
231                    && row.learned_evidence() == learned
232            })
233    }
234
235    pub(crate) fn has_learned_evidence(
236        &self,
237        sequence: u64,
238        learned: &ArtifactChecksumRecord,
239    ) -> bool {
240        self.stage_sequence == sequence
241            && self
242                .predecessors
243                .iter()
244                .any(|row| row.learned_evidence() == learned)
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.checksum(&self.workflow);
272        bytes.u64_be(self.stage_sequence);
273        bytes.checksum(&self.plan);
274        bytes.u64_be(self.predecessors.len() as u64);
275        for row in &self.predecessors {
276            bytes.u64_be(row.stage_sequence);
277            for digest in [&row.binding, &row.settlement, &row.learned_evidence] {
278                bytes.checksum(digest);
279            }
280        }
281        bytes.finish()
282    }
283}
284fn validate_child(
285    workflow: &ExecutionWorkflowRecord,
286    stage_sequence: u64,
287    plan: &OperationPlanRecord,
288) -> Result<(), ExecutionWorkflowError> {
289    let stage = workflow
290        .stage(stage_sequence)
291        .map_err(|_| ExecutionWorkflowError::UnknownStage(stage_sequence))?;
292    if plan.context() != workflow.allocation.context()
293        || plan.inventory() != workflow.allocation.inventory()
294        || plan.selected_targets() != [stage.target()]
295    {
296        return Err(ExecutionWorkflowError::ChildIdentityMismatch);
297    }
298    if plan.budget().mutations() != stage.budget().mutations()
299        || plan.budget().observations() != stage.budget().observations()
300    {
301        return Err(ExecutionWorkflowError::ChildBudgetMismatch);
302    }
303    if plan.operations().iter().any(|operation| {
304        operation.budget().mutations() == 0 && operation.budget().observations() == 0
305    }) {
306        return Err(ExecutionWorkflowError::UnallocatedOperation);
307    }
308    Ok(())
309}
310
311fn validate_predecessor_set(
312    workflow: &ExecutionWorkflowRecord,
313    stage_sequence: u64,
314    predecessors: &[ExecutionStagePredecessorRecord],
315) -> Result<(), ExecutionWorkflowError> {
316    let dependencies = workflow
317        .allocation
318        .graph()
319        .node(stage_sequence)
320        .map_err(|_| ExecutionWorkflowError::PredecessorMismatch)?
321        .depends_on();
322    if !predecessors
323        .iter()
324        .map(ExecutionStagePredecessorRecord::stage_sequence)
325        .eq(dependencies.iter().copied())
326    {
327        return Err(ExecutionWorkflowError::PredecessorMismatch);
328    }
329    Ok(())
330}
331
332fn canonical_predecessors(
333    rows: &mut [ExecutionStagePredecessorRecord],
334) -> Result<(), ExecutionWorkflowError> {
335    if rows.len() > MAX_EFFECT_DEPENDENCIES {
336        return Err(ExecutionWorkflowError::PredecessorMismatch);
337    }
338    rows.sort_by_key(ExecutionStagePredecessorRecord::stage_sequence);
339    if rows
340        .windows(2)
341        .any(|pair| pair[0].stage_sequence == pair[1].stage_sequence)
342    {
343        return Err(ExecutionWorkflowError::PredecessorMismatch);
344    }
345    Ok(())
346}
347fn bounded_predecessors<'de, D: Deserializer<'de>>(
348    deserializer: D,
349) -> Result<Vec<ExecutionStagePredecessorRecord>, D::Error> {
350    bounded_sequence::deserialize::<_, _, MAX_EFFECT_DEPENDENCIES>(
351        deserializer,
352        "at most 1,024 exact stage predecessors",
353        ExecutionWorkflowError::PredecessorMismatch,
354    )
355}
356/// Typed rejection without changing original records or spending.
357#[derive(Debug, Error, Eq, PartialEq)]
358pub enum ExecutionWorkflowError {
359    /// Only the distinct maintained v1 generation is admitted.
360    #[error("unsupported execution workflow/stage version {0}")]
361    UnsupportedVersion(u16),
362    /// Full context, inventory or exact single allocated target changed.
363    #[error("child plan differs from original stage identity")]
364    ChildIdentityMismatch,
365    /// Child aggregate ceilings differ from original assigned stage limits.
366    #[error("child plan differs from original stage budget")]
367    ChildBudgetMismatch,
368    /// A child operation has no assigned allowance.
369    #[error("child plan contains an unallocated operation")]
370    UnallocatedOperation,
371    /// Predecessors are excessive, duplicated, missing or outside the original graph.
372    #[error("stage predecessors differ from original dependencies")]
373    PredecessorMismatch,
374    /// Retained full workflow or child plan digest differs.
375    #[error("execution stage binding mismatch")]
376    BindingMismatch,
377    /// The original catalog has no stage with this identity.
378    #[error("unknown execution stage {0}")]
379    UnknownStage(u64),
380}
381
382#[cfg(test)]
383mod tests;
ic_backup/model/execution_workflow/mod.rs

ic_backup/model/execution_workflow/
mod.rs
