trellis_core/
oracle.rs

1use crate::output_payload::StoredOutput;
2use crate::{
3    FullRecomputeOutputMismatch, FullRecomputeResourceMismatch, Graph, GraphError, GraphResult,
4    NodeId, OutputKey, ResourceKey, ScopeId,
5};
6use std::collections::{BTreeMap, BTreeSet};
7
8/// Result of comparing incremental graph state against full recompute.
9#[derive(Clone, Debug, Eq, PartialEq)]
10pub struct FullRecomputeCheck {
11    /// Derived nodes checked in deterministic topological order.
12    pub checked_derived: Vec<NodeId>,
13    /// Collection nodes checked in deterministic topological order.
14    pub checked_collections: Vec<NodeId>,
15    /// Desired resource keys whose owner sets were checked.
16    pub checked_resources: Vec<ResourceKey>,
17    /// Materialized outputs whose current values were checked.
18    pub checked_outputs: Vec<OutputKey>,
19}
20
21impl<C: Clone + PartialEq> Graph<C> {
22    /// Recomputes supported graph state from canonical inputs and compares it.
23    pub fn full_recompute(&self) -> GraphResult<FullRecomputeCheck> {
24        self.full_recompute_check()
25    }
26
27    /// Asserts that incremental state equals a supported full recompute.
28    pub fn assert_incremental_equals_full(&self) -> GraphResult<FullRecomputeCheck> {
29        self.full_recompute_check()
30    }
31
32    /// Compares committed incremental state against full recompute.
33    pub fn full_recompute_check(&self) -> GraphResult<FullRecomputeCheck> {
34        let mut full = self.clone();
35        full.derived_values.clear();
36        full.collection_values.clear();
37        full.previous_collection_values.clear();
38        full.collection_diffs.clear();
39        full.resource_owners.clear();
40        full.resource_payloads.clear();
41        full.resource_acquisitions.clear();
42        full.next_resource_acquisition = 1;
43        full.output_values.clear();
44        let order = full.derived_topological_order()?;
45
46        for node in &order {
47            let dependencies = full
48                .nodes
49                .get(node)
50                .expect("derived node metadata exists")
51                .dependencies();
52            let value = full.compute_derived(*node, dependencies.as_slice())?;
53            full.derived_values.insert(*node, value);
54        }
55
56        for node in &order {
57            let incremental = self
58                .derived_values
59                .get(node)
60                .ok_or(GraphError::FullRecomputeMismatch(*node))?;
61            let recomputed = full
62                .derived_values
63                .get(node)
64                .ok_or(GraphError::FullRecomputeMismatch(*node))?;
65            if !incremental.equals(recomputed.as_ref()) {
66                return Err(GraphError::FullRecomputeMismatch(*node));
67            }
68        }
69
70        let collection_order = full.collection_topological_order()?;
71        let all_nodes: Vec<NodeId> = full.nodes.keys().copied().collect();
72        full.recompute_dirty_collections(&all_nodes)?;
73        self.compare_full_recomputed_collections(&full, &collection_order)?;
74        let checked_resources = self.compare_full_recomputed_resources(&mut full)?;
75        let checked_outputs = self.compare_full_recomputed_outputs(&mut full, &all_nodes)?;
76
77        Ok(FullRecomputeCheck {
78            checked_derived: order,
79            checked_collections: collection_order,
80            checked_resources,
81            checked_outputs,
82        })
83    }
84
85    fn compare_full_recomputed_resources(
86        &self,
87        full: &mut Graph<C>,
88    ) -> GraphResult<Vec<ResourceKey>> {
89        let planner_collections: Vec<NodeId> = full
90            .resource_planners
91            .iter()
92            .map(|planner| planner.collection)
93            .collect();
94        full.baseline_collection_diffs(&planner_collections);
95        full.produce_resource_plan(&[])?;
96        if let Some(mismatch) =
97            first_resource_owner_mismatch(&self.resource_owners, &full.resource_owners)
98        {
99            return Err(GraphError::FullRecomputeResourceMismatch(mismatch));
100        }
101        Ok(self.resource_owners.keys().cloned().collect())
102    }
103
104    fn compare_full_recomputed_outputs(
105        &self,
106        full: &mut Graph<C>,
107        all_nodes: &[NodeId],
108    ) -> GraphResult<Vec<OutputKey>> {
109        full.produce_output_frames(
110            all_nodes,
111            &[],
112            &BTreeMap::new(),
113            self.next_transaction_id,
114            self.revision,
115        )?;
116        if let Some(mismatch) =
117            first_output_value_mismatch(&self.output_values, &full.output_values)
118        {
119            return Err(GraphError::FullRecomputeOutputMismatch(mismatch));
120        }
121        Ok(self.output_values.keys().copied().collect())
122    }
123}
124
125fn first_resource_owner_mismatch(
126    incremental: &BTreeMap<ResourceKey, BTreeSet<ScopeId>>,
127    recomputed: &BTreeMap<ResourceKey, BTreeSet<ScopeId>>,
128) -> Option<FullRecomputeResourceMismatch> {
129    let keys: BTreeSet<ResourceKey> = incremental
130        .keys()
131        .chain(recomputed.keys())
132        .cloned()
133        .collect();
134    for key in keys {
135        let incremental_owners = owner_vec(incremental.get(&key));
136        let recomputed_owners = owner_vec(recomputed.get(&key));
137        if incremental_owners != recomputed_owners {
138            return Some(FullRecomputeResourceMismatch {
139                key,
140                incremental_owners,
141                recomputed_owners,
142            });
143        }
144    }
145    None
146}
147
148fn owner_vec(owners: Option<&BTreeSet<ScopeId>>) -> Vec<ScopeId> {
149    owners
150        .into_iter()
151        .flat_map(|owners| owners.iter().copied())
152        .collect()
153}
154
155fn first_output_value_mismatch(
156    incremental: &BTreeMap<OutputKey, Box<dyn StoredOutput>>,
157    recomputed: &BTreeMap<OutputKey, Box<dyn StoredOutput>>,
158) -> Option<FullRecomputeOutputMismatch> {
159    let keys: BTreeSet<OutputKey> = incremental
160        .keys()
161        .chain(recomputed.keys())
162        .copied()
163        .collect();
164    for key in keys {
165        let incremental_value = incremental.get(&key);
166        let recomputed_value = recomputed.get(&key);
167        let matches = match (incremental_value, recomputed_value) {
168            (Some(incremental), Some(recomputed)) => incremental.equals(recomputed.as_ref()),
169            (None, None) => true,
170            _ => false,
171        };
172        if !matches {
173            return Some(FullRecomputeOutputMismatch {
174                key,
175                incremental_present: incremental_value.is_some(),
176                recomputed_present: recomputed_value.is_some(),
177            });
178        }
179    }
180    None
181}
trellis_core/oracle.rs

trellis_core/
oracle.rs
