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> 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.output_values.clear();
41        let order = full.derived_topological_order()?;
42
43        for node in &order {
44            let dependencies = full
45                .nodes
46                .get(node)
47                .expect("derived node metadata exists")
48                .dependencies();
49            let value = full.compute_derived(*node, dependencies.as_slice())?;
50            full.derived_values.insert(*node, value);
51        }
52
53        for node in &order {
54            let incremental = self
55                .derived_values
56                .get(node)
57                .ok_or(GraphError::FullRecomputeMismatch(*node))?;
58            let recomputed = full
59                .derived_values
60                .get(node)
61                .ok_or(GraphError::FullRecomputeMismatch(*node))?;
62            if !incremental.equals(recomputed.as_ref()) {
63                return Err(GraphError::FullRecomputeMismatch(*node));
64            }
65        }
66
67        let collection_order = full.collection_topological_order()?;
68        let all_nodes: Vec<NodeId> = full.nodes.keys().copied().collect();
69        full.recompute_dirty_collections(&all_nodes)?;
70        self.compare_full_recomputed_collections(&full, &collection_order)?;
71        let checked_resources = self.compare_full_recomputed_resources(&mut full)?;
72        let checked_outputs = self.compare_full_recomputed_outputs(&mut full, &all_nodes)?;
73
74        Ok(FullRecomputeCheck {
75            checked_derived: order,
76            checked_collections: collection_order,
77            checked_resources,
78            checked_outputs,
79        })
80    }
81
82    fn compare_full_recomputed_resources(
83        &self,
84        full: &mut Graph<C>,
85    ) -> GraphResult<Vec<ResourceKey>> {
86        let planner_collections: Vec<NodeId> = full
87            .resource_planners
88            .iter()
89            .map(|planner| planner.collection)
90            .collect();
91        full.baseline_collection_diffs(&planner_collections);
92        full.produce_resource_plan(&[])?;
93        if let Some(mismatch) =
94            first_resource_owner_mismatch(&self.resource_owners, &full.resource_owners)
95        {
96            return Err(GraphError::FullRecomputeResourceMismatch(mismatch));
97        }
98        Ok(self.resource_owners.keys().cloned().collect())
99    }
100
101    fn compare_full_recomputed_outputs(
102        &self,
103        full: &mut Graph<C>,
104        all_nodes: &[NodeId],
105    ) -> GraphResult<Vec<OutputKey>> {
106        full.produce_output_frames(
107            all_nodes,
108            &[],
109            &BTreeMap::new(),
110            self.next_transaction_id,
111            self.revision,
112        )?;
113        if let Some(mismatch) =
114            first_output_value_mismatch(&self.output_values, &full.output_values)
115        {
116            return Err(GraphError::FullRecomputeOutputMismatch(mismatch));
117        }
118        Ok(self.output_values.keys().copied().collect())
119    }
120}
121
122fn first_resource_owner_mismatch(
123    incremental: &BTreeMap<ResourceKey, BTreeSet<ScopeId>>,
124    recomputed: &BTreeMap<ResourceKey, BTreeSet<ScopeId>>,
125) -> Option<FullRecomputeResourceMismatch> {
126    let keys: BTreeSet<ResourceKey> = incremental
127        .keys()
128        .chain(recomputed.keys())
129        .cloned()
130        .collect();
131    for key in keys {
132        let incremental_owners = owner_vec(incremental.get(&key));
133        let recomputed_owners = owner_vec(recomputed.get(&key));
134        if incremental_owners != recomputed_owners {
135            return Some(FullRecomputeResourceMismatch {
136                key,
137                incremental_owners,
138                recomputed_owners,
139            });
140        }
141    }
142    None
143}
144
145fn owner_vec(owners: Option<&BTreeSet<ScopeId>>) -> Vec<ScopeId> {
146    owners
147        .into_iter()
148        .flat_map(|owners| owners.iter().copied())
149        .collect()
150}
151
152fn first_output_value_mismatch(
153    incremental: &BTreeMap<OutputKey, Box<dyn StoredOutput>>,
154    recomputed: &BTreeMap<OutputKey, Box<dyn StoredOutput>>,
155) -> Option<FullRecomputeOutputMismatch> {
156    let keys: BTreeSet<OutputKey> = incremental
157        .keys()
158        .chain(recomputed.keys())
159        .copied()
160        .collect();
161    for key in keys {
162        let incremental_value = incremental.get(&key);
163        let recomputed_value = recomputed.get(&key);
164        let matches = match (incremental_value, recomputed_value) {
165            (Some(incremental), Some(recomputed)) => incremental.equals(recomputed.as_ref()),
166            (None, None) => true,
167            _ => false,
168        };
169        if !matches {
170            return Some(FullRecomputeOutputMismatch {
171                key,
172                incremental_present: incremental_value.is_some(),
173                recomputed_present: recomputed_value.is_some(),
174            });
175        }
176    }
177    None
178}
trellis_core/oracle.rs

trellis_core/
oracle.rs
