omena_reactive/
graph.rs

1use std::{
2    collections::BTreeSet,
3    error::Error,
4    fmt,
5    sync::atomic::{AtomicU64, Ordering},
6};
7
8use crate::{ChangePolicyV0, ReactiveEngineV0, ReactiveStateV0};
9
10static NEXT_REACTIVE_GRAPH_ID_V0: AtomicU64 = AtomicU64::new(1);
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
13pub(crate) struct ReactiveGraphIdV0(u64);
14
15impl ReactiveGraphIdV0 {
16    fn fresh() -> Self {
17        let id = match NEXT_REACTIVE_GRAPH_ID_V0.fetch_update(
18            Ordering::Relaxed,
19            Ordering::Relaxed,
20            |next| next.checked_add(1),
21        ) {
22            Ok(id) => id,
23            Err(_) => std::process::abort(),
24        };
25        Self(id)
26    }
27
28    pub(crate) fn value(self) -> u64 {
29        self.0
30    }
31}
32
33/// An append-only node index branded by the builder that minted it.
34///
35/// `Ord` is intentionally derived with `index` declared first, so iteration
36/// order within one graph remains insertion order. Builders expose no removal
37/// or recycling operation and `build(self)` consumes the builder, so node
38/// generations are deliberately not represented.
39#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
40pub struct ReactiveNodeIdV0 {
41    pub(crate) index: usize,
42    pub(crate) graph: ReactiveGraphIdV0,
43}
44
45impl ReactiveNodeIdV0 {
46    pub fn index(self) -> usize {
47        self.index
48    }
49}
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
52#[non_exhaustive]
53pub enum ReactiveNodeKindV0 {
54    Input,
55    Map,
56    Zip,
57    Switch,
58    DeltaFold,
59    AsyncResult,
60    EffectBoundary,
61}
62
63pub type MapOperationV0 = fn(&ReactiveStateV0) -> ReactiveStateV0;
64pub type ZipOperationV0 = fn(&ReactiveStateV0, &ReactiveStateV0) -> ReactiveStateV0;
65
66#[derive(Clone)]
67pub(crate) enum NodeOperationV0 {
68    Input,
69    Map { operation: MapOperationV0 },
70    Zip { operation: ZipOperationV0 },
71    Switch,
72    DeltaFold { keys: Vec<String> },
73    AsyncResult,
74    EffectBoundary { channel: String },
75}
76
77impl NodeOperationV0 {
78    pub(crate) fn kind(&self) -> ReactiveNodeKindV0 {
79        match self {
80            Self::Input => ReactiveNodeKindV0::Input,
81            Self::Map { .. } => ReactiveNodeKindV0::Map,
82            Self::Zip { .. } => ReactiveNodeKindV0::Zip,
83            Self::Switch => ReactiveNodeKindV0::Switch,
84            Self::DeltaFold { .. } => ReactiveNodeKindV0::DeltaFold,
85            Self::AsyncResult => ReactiveNodeKindV0::AsyncResult,
86            Self::EffectBoundary { .. } => ReactiveNodeKindV0::EffectBoundary,
87        }
88    }
89}
90
91#[derive(Clone)]
92pub(crate) struct NodeBlueprintV0 {
93    pub(crate) operation: NodeOperationV0,
94    pub(crate) dependencies: Vec<ReactiveNodeIdV0>,
95    pub(crate) height: u32,
96    pub(crate) initial_state: ReactiveStateV0,
97    pub(crate) change_policy: ChangePolicyV0,
98}
99
100#[derive(Debug, Clone, PartialEq, Eq)]
101#[non_exhaustive]
102pub enum ReactiveGraphBuildErrorV0 {
103    #[non_exhaustive]
104    EmptyChangePolicyName { node_index: usize },
105    #[non_exhaustive]
106    DuplicateDeltaKey { key: String },
107    #[non_exhaustive]
108    ForeignNodeId {
109        node_index: usize,
110        expected_graph: u64,
111        actual_graph: u64,
112    },
113}
114
115impl fmt::Display for ReactiveGraphBuildErrorV0 {
116    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
117        match self {
118            Self::EmptyChangePolicyName { node_index } => {
119                write!(
120                    formatter,
121                    "node {node_index} has an empty change-policy name"
122                )
123            }
124            Self::DuplicateDeltaKey { key } => {
125                write!(formatter, "delta-fold key `{key}` is duplicated")
126            }
127            Self::ForeignNodeId {
128                node_index,
129                expected_graph,
130                actual_graph,
131            } => write!(
132                formatter,
133                "node {node_index} belongs to reactive graph {actual_graph}, not {expected_graph}"
134            ),
135        }
136    }
137}
138
139impl Error for ReactiveGraphBuildErrorV0 {}
140
141pub struct ReactiveGraphBuilderV0 {
142    graph_id: ReactiveGraphIdV0,
143    nodes: Vec<NodeBlueprintV0>,
144    foreign_dependency: Option<(usize, ReactiveGraphIdV0)>,
145}
146
147impl Default for ReactiveGraphBuilderV0 {
148    fn default() -> Self {
149        Self::new()
150    }
151}
152
153impl ReactiveGraphBuilderV0 {
154    pub fn new() -> Self {
155        Self {
156            graph_id: ReactiveGraphIdV0::fresh(),
157            nodes: Vec::new(),
158            foreign_dependency: None,
159        }
160    }
161
162    pub fn add_input(
163        &mut self,
164        initial_state: ReactiveStateV0,
165        change_policy: ChangePolicyV0,
166    ) -> ReactiveNodeIdV0 {
167        self.push(
168            NodeOperationV0::Input,
169            Vec::new(),
170            initial_state,
171            change_policy,
172        )
173    }
174
175    pub fn add_async_result(
176        &mut self,
177        initial_state: ReactiveStateV0,
178        change_policy: ChangePolicyV0,
179    ) -> ReactiveNodeIdV0 {
180        self.push(
181            NodeOperationV0::AsyncResult,
182            Vec::new(),
183            initial_state,
184            change_policy,
185        )
186    }
187
188    pub fn add_map(
189        &mut self,
190        dependency: ReactiveNodeIdV0,
191        operation: MapOperationV0,
192        change_policy: ChangePolicyV0,
193    ) -> ReactiveNodeIdV0 {
194        self.push(
195            NodeOperationV0::Map { operation },
196            vec![dependency],
197            ReactiveStateV0::pending(),
198            change_policy,
199        )
200    }
201
202    pub fn add_zip(
203        &mut self,
204        left: ReactiveNodeIdV0,
205        right: ReactiveNodeIdV0,
206        operation: ZipOperationV0,
207        change_policy: ChangePolicyV0,
208    ) -> ReactiveNodeIdV0 {
209        self.push(
210            NodeOperationV0::Zip { operation },
211            vec![left, right],
212            ReactiveStateV0::pending(),
213            change_policy,
214        )
215    }
216
217    /// Adds a static two-branch switch. Branch identities cannot be changed
218    /// after the graph is built.
219    pub fn add_switch(
220        &mut self,
221        selector: ReactiveNodeIdV0,
222        when_false: ReactiveNodeIdV0,
223        when_true: ReactiveNodeIdV0,
224        change_policy: ChangePolicyV0,
225    ) -> ReactiveNodeIdV0 {
226        self.push(
227            NodeOperationV0::Switch,
228            vec![selector, when_false, when_true],
229            ReactiveStateV0::pending(),
230            change_policy,
231        )
232    }
233
234    pub fn add_delta_fold(
235        &mut self,
236        entries: Vec<(String, ReactiveNodeIdV0)>,
237        change_policy: ChangePolicyV0,
238    ) -> Result<ReactiveNodeIdV0, ReactiveGraphBuildErrorV0> {
239        if let Some((_, dependency)) = entries
240            .iter()
241            .find(|(_, dependency)| dependency.graph != self.graph_id)
242        {
243            return Err(ReactiveGraphBuildErrorV0::ForeignNodeId {
244                node_index: self.nodes.len(),
245                expected_graph: self.graph_id.value(),
246                actual_graph: dependency.graph.value(),
247            });
248        }
249        let mut seen = BTreeSet::new();
250        for (key, _) in &entries {
251            if !seen.insert(key.clone()) {
252                return Err(ReactiveGraphBuildErrorV0::DuplicateDeltaKey { key: key.clone() });
253            }
254        }
255        let (keys, dependencies): (Vec<_>, Vec<_>) = entries.into_iter().unzip();
256        Ok(self.push(
257            NodeOperationV0::DeltaFold { keys },
258            dependencies,
259            ReactiveStateV0::pending(),
260            change_policy,
261        ))
262    }
263
264    pub fn add_effect_boundary(
265        &mut self,
266        dependency: ReactiveNodeIdV0,
267        channel: impl Into<String>,
268        change_policy: ChangePolicyV0,
269    ) -> ReactiveNodeIdV0 {
270        self.push(
271            NodeOperationV0::EffectBoundary {
272                channel: channel.into(),
273            },
274            vec![dependency],
275            ReactiveStateV0::pending(),
276            change_policy,
277        )
278    }
279
280    pub fn build(self) -> Result<ReactiveEngineV0, ReactiveGraphBuildErrorV0> {
281        let Self {
282            graph_id,
283            nodes,
284            foreign_dependency,
285        } = self;
286        if let Some((node_index, actual_graph)) = foreign_dependency {
287            return Err(ReactiveGraphBuildErrorV0::ForeignNodeId {
288                node_index,
289                expected_graph: graph_id.value(),
290                actual_graph: actual_graph.value(),
291            });
292        }
293        for (node_index, node) in nodes.iter().enumerate() {
294            if node.change_policy.name().is_empty() {
295                return Err(ReactiveGraphBuildErrorV0::EmptyChangePolicyName { node_index });
296            }
297        }
298        Ok(ReactiveEngineV0::from_blueprints(graph_id, nodes))
299    }
300
301    fn push(
302        &mut self,
303        operation: NodeOperationV0,
304        dependencies: Vec<ReactiveNodeIdV0>,
305        initial_state: ReactiveStateV0,
306        change_policy: ChangePolicyV0,
307    ) -> ReactiveNodeIdV0 {
308        let node_index = self.nodes.len();
309        if self.foreign_dependency.is_none() {
310            self.foreign_dependency = dependencies
311                .iter()
312                .find(|dependency| dependency.graph != self.graph_id)
313                .map(|dependency| (node_index, dependency.graph));
314        }
315        debug_assert!(
316            dependencies.iter().all(|dependency| {
317                dependency.graph != self.graph_id || dependency.index() < self.nodes.len()
318            }),
319            "same-graph dependencies must refer to earlier nodes; graph ownership is checked separately"
320        );
321        let height = dependencies
322            .iter()
323            .filter(|dependency| dependency.graph == self.graph_id)
324            .filter_map(|dependency| self.nodes.get(dependency.index()))
325            .map(|dependency| dependency.height)
326            .max()
327            .map_or(0, |height| height.saturating_add(1));
328        let id = ReactiveNodeIdV0 {
329            index: node_index,
330            graph: self.graph_id,
331        };
332        self.nodes.push(NodeBlueprintV0 {
333            operation,
334            dependencies,
335            height,
336            initial_state,
337            change_policy,
338        });
339        id
340    }
341}
342
343#[cfg(test)]
344mod tests {
345    use std::collections::BTreeSet;
346
347    use super::*;
348    use crate::ReactiveValueV0;
349
350    fn exact_policy() -> ChangePolicyV0 {
351        ChangePolicyV0::exact("graphOwnershipTestValue")
352    }
353
354    fn counter(value: u64) -> ReactiveStateV0 {
355        ReactiveStateV0::available(ReactiveValueV0::Counter(value))
356    }
357
358    fn identity(state: &ReactiveStateV0) -> ReactiveStateV0 {
359        state.clone()
360    }
361
362    #[test]
363    fn graph_build_rejects_a_dependency_from_another_builder() {
364        let mut first = ReactiveGraphBuilderV0::new();
365        first.add_input(counter(1), exact_policy());
366
367        let mut second = ReactiveGraphBuilderV0::new();
368        let foreign = second.add_input(counter(2), exact_policy());
369
370        first.add_map(foreign, identity, exact_policy());
371        assert!(matches!(
372            first.build(),
373            Err(ReactiveGraphBuildErrorV0::ForeignNodeId {
374                node_index: 1,
375                expected_graph,
376                actual_graph,
377            }) if expected_graph != actual_graph
378        ));
379    }
380
381    #[test]
382    fn structurally_identical_graphs_still_have_distinct_identities() {
383        let mut first = ReactiveGraphBuilderV0::new();
384        first.add_input(counter(1), exact_policy());
385
386        let mut second = ReactiveGraphBuilderV0::new();
387        let foreign = second.add_input(counter(1), exact_policy());
388
389        first.add_map(foreign, identity, exact_policy());
390        assert!(matches!(
391            first.build(),
392            Err(ReactiveGraphBuildErrorV0::ForeignNodeId {
393                node_index: 1,
394                expected_graph,
395                actual_graph,
396            }) if expected_graph != actual_graph
397        ));
398    }
399
400    #[test]
401    fn delta_fold_rejects_a_foreign_dependency_immediately() {
402        let mut first = ReactiveGraphBuilderV0::new();
403        first.add_input(counter(1), exact_policy());
404
405        let mut second = ReactiveGraphBuilderV0::new();
406        let foreign = second.add_input(counter(2), exact_policy());
407
408        assert!(matches!(
409            first.add_delta_fold(vec![("foreign".to_string(), foreign)], exact_policy()),
410            Err(ReactiveGraphBuildErrorV0::ForeignNodeId {
411                node_index: 1,
412                expected_graph,
413                actual_graph,
414            }) if expected_graph != actual_graph
415        ));
416    }
417
418    #[test]
419    fn derived_node_order_preserves_insertion_order_within_one_graph() {
420        let mut graph = ReactiveGraphBuilderV0::new();
421        let ids = (0..4)
422            .map(|value| graph.add_input(counter(value), exact_policy()))
423            .collect::<BTreeSet<_>>();
424
425        assert_eq!(
426            ids.into_iter()
427                .map(ReactiveNodeIdV0::index)
428                .collect::<Vec<_>>(),
429            vec![0, 1, 2, 3]
430        );
431    }
432}
omena_reactive/graph.rs

omena_reactive/
graph.rs
