omena_reactive/
engine.rs

1use std::{
2    collections::{BTreeMap, BTreeSet, VecDeque},
3    error::Error,
4    fmt,
5};
6
7use crate::{
8    ChangePolicyV0, ReactiveNodeIdV0, ReactiveNodeKindV0, ReactiveStateV0, ReactiveUnavailableV0,
9    ReactiveValueV0,
10    graph::{NodeBlueprintV0, NodeOperationV0, ReactiveGraphIdV0},
11};
12
13#[derive(Debug, Clone, PartialEq, Eq)]
14#[non_exhaustive]
15pub struct EffectReceiptV0 {
16    pub channel: String,
17    pub wave: u64,
18    pub state: ReactiveStateV0,
19}
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22#[non_exhaustive]
23pub enum StabilizeStatusV0 {
24    #[non_exhaustive]
25    Settled {
26        wave: u64,
27        recomputed_node_count: usize,
28    },
29    #[non_exhaustive]
30    Pending {
31        wave: u64,
32        recomputed_node_count: usize,
33    },
34}
35
36#[derive(Debug, Clone, PartialEq, Eq)]
37#[non_exhaustive]
38pub enum ReactiveEngineErrorV0 {
39    #[non_exhaustive]
40    InvalidNode {
41        node_index: usize,
42    },
43    #[non_exhaustive]
44    ForeignNodeId {
45        node_index: usize,
46        expected_graph: u64,
47        actual_graph: u64,
48    },
49    #[non_exhaustive]
50    NodeDoesNotAcceptDeposits {
51        node_index: usize,
52    },
53    ObserverMutationDuringWave,
54    ZeroStepBudget,
55}
56
57impl fmt::Display for ReactiveEngineErrorV0 {
58    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
59        match self {
60            Self::InvalidNode { node_index } => {
61                write!(formatter, "reactive node {node_index} does not exist")
62            }
63            Self::ForeignNodeId {
64                node_index,
65                expected_graph,
66                actual_graph,
67            } => write!(
68                formatter,
69                "reactive node {node_index} belongs to reactive graph {actual_graph}, not {expected_graph}"
70            ),
71            Self::NodeDoesNotAcceptDeposits { node_index } => {
72                write!(
73                    formatter,
74                    "reactive node {node_index} does not accept deposits"
75                )
76            }
77            Self::ObserverMutationDuringWave => {
78                write!(
79                    formatter,
80                    "observers cannot be changed during stabilization"
81                )
82            }
83            Self::ZeroStepBudget => write!(formatter, "stabilization step budget must be non-zero"),
84        }
85    }
86}
87
88impl Error for ReactiveEngineErrorV0 {}
89
90#[derive(Debug, Clone, PartialEq, Eq)]
91#[non_exhaustive]
92pub enum DeltaFoldParityErrorV0 {
93    #[non_exhaustive]
94    InvalidNode { node_index: usize },
95    #[non_exhaustive]
96    ForeignNodeId {
97        node_index: usize,
98        expected_graph: u64,
99        actual_graph: u64,
100    },
101    #[non_exhaustive]
102    NotDeltaFold { node: ReactiveNodeIdV0 },
103    #[non_exhaustive]
104    Diverged {
105        node: ReactiveNodeIdV0,
106        incremental_digest: [u8; 32],
107        rebuilt_digest: [u8; 32],
108    },
109}
110
111impl fmt::Display for DeltaFoldParityErrorV0 {
112    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
113        match self {
114            Self::InvalidNode { node_index } => {
115                write!(formatter, "reactive node {node_index} does not exist")
116            }
117            Self::ForeignNodeId {
118                node_index,
119                expected_graph,
120                actual_graph,
121            } => write!(
122                formatter,
123                "reactive node {node_index} belongs to reactive graph {actual_graph}, not {expected_graph}"
124            ),
125            Self::NotDeltaFold { node } => {
126                write!(
127                    formatter,
128                    "reactive node {} is not a delta fold",
129                    node.index()
130                )
131            }
132            Self::Diverged { node, .. } => write!(
133                formatter,
134                "delta-fold node {} diverged from a full rebuild",
135                node.index()
136            ),
137        }
138    }
139}
140
141impl Error for DeltaFoldParityErrorV0 {}
142
143struct RuntimeNodeV0 {
144    operation: NodeOperationV0,
145    dependencies: Vec<ReactiveNodeIdV0>,
146    parents: Vec<ReactiveNodeIdV0>,
147    height: u32,
148    state: ReactiveStateV0,
149    change_policy: ChangePolicyV0,
150    observer_count: usize,
151    needed_by: usize,
152    stale: bool,
153    scheduled_wave: Option<u64>,
154    last_recomputed_wave: Option<u64>,
155    recompute_count: usize,
156    delta_entries: BTreeMap<String, ReactiveStateV0>,
157    delta_dirty_dependencies: BTreeSet<ReactiveNodeIdV0>,
158    delta_update_count: usize,
159}
160
161pub struct ReactiveEngineV0 {
162    graph_id: ReactiveGraphIdV0,
163    nodes: Vec<RuntimeNodeV0>,
164    deferred_deposits: BTreeMap<ReactiveNodeIdV0, ReactiveStateV0>,
165    activation_nodes: BTreeSet<ReactiveNodeIdV0>,
166    queues: BTreeMap<u32, VecDeque<ReactiveNodeIdV0>>,
167    effect_receipts: Vec<EffectReceiptV0>,
168    current_wave: u64,
169    current_wave_recomputes: usize,
170    stabilizing: bool,
171}
172
173impl ReactiveEngineV0 {
174    pub(crate) fn from_blueprints(
175        graph_id: ReactiveGraphIdV0,
176        blueprints: Vec<NodeBlueprintV0>,
177    ) -> Self {
178        let mut nodes: Vec<_> = blueprints
179            .into_iter()
180            .map(|blueprint| {
181                let stale = !matches!(
182                    blueprint.operation,
183                    NodeOperationV0::Input | NodeOperationV0::AsyncResult
184                );
185                RuntimeNodeV0 {
186                    operation: blueprint.operation,
187                    dependencies: blueprint.dependencies,
188                    parents: Vec::new(),
189                    height: blueprint.height,
190                    state: blueprint.initial_state,
191                    change_policy: blueprint.change_policy,
192                    observer_count: 0,
193                    needed_by: 0,
194                    stale,
195                    scheduled_wave: None,
196                    last_recomputed_wave: None,
197                    recompute_count: 0,
198                    delta_entries: BTreeMap::new(),
199                    delta_dirty_dependencies: BTreeSet::new(),
200                    delta_update_count: 0,
201                }
202            })
203            .collect();
204
205        for node_index in 0..nodes.len() {
206            let parent = ReactiveNodeIdV0 {
207                index: node_index,
208                graph: graph_id,
209            };
210            for dependency in nodes[node_index].dependencies.clone() {
211                nodes[dependency.index()].parents.push(parent);
212            }
213        }
214
215        Self {
216            graph_id,
217            nodes,
218            deferred_deposits: BTreeMap::new(),
219            activation_nodes: BTreeSet::new(),
220            queues: BTreeMap::new(),
221            effect_receipts: Vec::new(),
222            current_wave: 0,
223            current_wave_recomputes: 0,
224            stabilizing: false,
225        }
226    }
227
228    pub fn node_count(&self) -> usize {
229        self.nodes.len()
230    }
231
232    pub fn node_kind(
233        &self,
234        node: ReactiveNodeIdV0,
235    ) -> Result<ReactiveNodeKindV0, ReactiveEngineErrorV0> {
236        Ok(self.node(node)?.operation.kind())
237    }
238
239    pub fn state(&self, node: ReactiveNodeIdV0) -> Result<&ReactiveStateV0, ReactiveEngineErrorV0> {
240        Ok(&self.node(node)?.state)
241    }
242
243    pub fn is_necessary(&self, node: ReactiveNodeIdV0) -> Result<bool, ReactiveEngineErrorV0> {
244        Ok(self.node(node)?.needed_by > 0)
245    }
246
247    pub fn is_stale(&self, node: ReactiveNodeIdV0) -> Result<bool, ReactiveEngineErrorV0> {
248        Ok(self.node(node)?.stale)
249    }
250
251    pub fn node_recompute_count(
252        &self,
253        node: ReactiveNodeIdV0,
254    ) -> Result<usize, ReactiveEngineErrorV0> {
255        Ok(self.node(node)?.recompute_count)
256    }
257
258    pub fn delta_update_count(
259        &self,
260        node: ReactiveNodeIdV0,
261    ) -> Result<usize, ReactiveEngineErrorV0> {
262        Ok(self.node(node)?.delta_update_count)
263    }
264
265    pub fn current_wave(&self) -> u64 {
266        self.current_wave
267    }
268
269    pub fn is_stabilizing(&self) -> bool {
270        self.stabilizing
271    }
272
273    pub fn has_pending_work(&self) -> bool {
274        self.stabilizing || !self.deferred_deposits.is_empty() || !self.activation_nodes.is_empty()
275    }
276
277    /// Queues the latest value for an input-like node. The deposit is applied
278    /// only when the next wave begins, including when it arrives between
279    /// bounded steps of the current wave.
280    pub fn deposit(
281        &mut self,
282        node: ReactiveNodeIdV0,
283        state: ReactiveStateV0,
284    ) -> Result<(), ReactiveEngineErrorV0> {
285        match self.node(node)?.operation {
286            NodeOperationV0::Input | NodeOperationV0::AsyncResult => {
287                self.deferred_deposits.insert(node, state);
288                Ok(())
289            }
290            _ => Err(ReactiveEngineErrorV0::NodeDoesNotAcceptDeposits {
291                node_index: node.index(),
292            }),
293        }
294    }
295
296    pub fn observe(&mut self, node: ReactiveNodeIdV0) -> Result<(), ReactiveEngineErrorV0> {
297        if self.stabilizing {
298            return Err(ReactiveEngineErrorV0::ObserverMutationDuringWave);
299        }
300        self.node(node)?;
301        self.nodes[node.index()].observer_count += 1;
302        if self.nodes[node.index()].observer_count == 1 {
303            self.increase_necessity(node);
304        }
305        Ok(())
306    }
307
308    pub fn unobserve(&mut self, node: ReactiveNodeIdV0) -> Result<(), ReactiveEngineErrorV0> {
309        if self.stabilizing {
310            return Err(ReactiveEngineErrorV0::ObserverMutationDuringWave);
311        }
312        self.node(node)?;
313        if self.nodes[node.index()].observer_count == 0 {
314            return Ok(());
315        }
316        self.nodes[node.index()].observer_count -= 1;
317        if self.nodes[node.index()].observer_count == 0 {
318            self.decrease_necessity(node);
319        }
320        Ok(())
321    }
322
323    pub fn stabilize_step(
324        &mut self,
325        maximum_recomputes: usize,
326    ) -> Result<StabilizeStatusV0, ReactiveEngineErrorV0> {
327        if maximum_recomputes == 0 {
328            return Err(ReactiveEngineErrorV0::ZeroStepBudget);
329        }
330        if !self.stabilizing {
331            self.begin_wave();
332        }
333        if !self.stabilizing {
334            return Ok(StabilizeStatusV0::Settled {
335                wave: self.current_wave,
336                recomputed_node_count: 0,
337            });
338        }
339
340        let mut recomputed = 0;
341        while recomputed < maximum_recomputes {
342            let Some(node) = self.pop_next_scheduled() else {
343                self.stabilizing = false;
344                return Ok(StabilizeStatusV0::Settled {
345                    wave: self.current_wave,
346                    recomputed_node_count: recomputed,
347                });
348            };
349            if self.nodes[node.index()].needed_by == 0 {
350                self.nodes[node.index()].stale = true;
351                continue;
352            }
353            self.recompute(node);
354            recomputed += 1;
355            self.current_wave_recomputes += 1;
356        }
357
358        if self.queues.values().all(VecDeque::is_empty) {
359            self.stabilizing = false;
360            Ok(StabilizeStatusV0::Settled {
361                wave: self.current_wave,
362                recomputed_node_count: recomputed,
363            })
364        } else {
365            Ok(StabilizeStatusV0::Pending {
366                wave: self.current_wave,
367                recomputed_node_count: recomputed,
368            })
369        }
370    }
371
372    pub fn stabilize_until_settled(
373        &mut self,
374        maximum_recomputes: usize,
375    ) -> Result<StabilizeStatusV0, ReactiveEngineErrorV0> {
376        if maximum_recomputes == 0 {
377            return Err(ReactiveEngineErrorV0::ZeroStepBudget);
378        }
379        let mut remaining = maximum_recomputes;
380        loop {
381            let status = self.stabilize_step(remaining)?;
382            match status {
383                StabilizeStatusV0::Settled { .. } => return Ok(status),
384                StabilizeStatusV0::Pending {
385                    recomputed_node_count,
386                    ..
387                } => {
388                    remaining = remaining.saturating_sub(recomputed_node_count);
389                    if remaining == 0 {
390                        return Ok(status);
391                    }
392                }
393            }
394        }
395    }
396
397    pub fn drain_effect_receipts(&mut self) -> Vec<EffectReceiptV0> {
398        std::mem::take(&mut self.effect_receipts)
399    }
400
401    pub fn verify_delta_fold(&self, node: ReactiveNodeIdV0) -> Result<(), DeltaFoldParityErrorV0> {
402        if node.graph != self.graph_id {
403            return Err(DeltaFoldParityErrorV0::ForeignNodeId {
404                node_index: node.index(),
405                expected_graph: self.graph_id.value(),
406                actual_graph: node.graph.value(),
407            });
408        }
409        let Some(runtime) = self.nodes.get(node.index()) else {
410            return Err(DeltaFoldParityErrorV0::InvalidNode {
411                node_index: node.index(),
412            });
413        };
414        if !matches!(runtime.operation, NodeOperationV0::DeltaFold { .. }) {
415            return Err(DeltaFoldParityErrorV0::NotDeltaFold { node });
416        }
417        let ReactiveStateV0::Available(ReactiveValueV0::Digest(incremental_digest)) =
418            &runtime.state
419        else {
420            return Ok(());
421        };
422        let Some(rebuilt_entries) = self.rebuild_delta_fold_from_dependencies(node) else {
423            return Err(DeltaFoldParityErrorV0::NotDeltaFold { node });
424        };
425        let rebuilt_digest = full_delta_digest(&rebuilt_entries);
426        if *incremental_digest == rebuilt_digest && runtime.delta_entries == rebuilt_entries {
427            Ok(())
428        } else {
429            Err(DeltaFoldParityErrorV0::Diverged {
430                node,
431                incremental_digest: *incremental_digest,
432                rebuilt_digest,
433            })
434        }
435    }
436
437    fn rebuild_delta_fold_from_dependencies(
438        &self,
439        node: ReactiveNodeIdV0,
440    ) -> Option<BTreeMap<String, ReactiveStateV0>> {
441        let runtime = self.nodes.get(node.index())?;
442        let NodeOperationV0::DeltaFold { keys } = &runtime.operation else {
443            return None;
444        };
445        Some(
446            keys.iter()
447                .zip(&runtime.dependencies)
448                .map(|(key, dependency)| {
449                    (key.clone(), self.nodes[dependency.index()].state.clone())
450                })
451                .collect(),
452        )
453    }
454
455    fn begin_wave(&mut self) {
456        if self.deferred_deposits.is_empty() && self.activation_nodes.is_empty() {
457            return;
458        }
459        self.current_wave = self.current_wave.saturating_add(1);
460        self.current_wave_recomputes = 0;
461        self.stabilizing = true;
462
463        let deposits = std::mem::take(&mut self.deferred_deposits);
464        for (node, next) in deposits {
465            let previous = self.nodes[node.index()].state.clone();
466            let policy = self.nodes[node.index()].change_policy;
467            self.nodes[node.index()].state = next.clone();
468            self.nodes[node.index()].stale = false;
469            if !policy.equivalent(&previous, &next) {
470                self.schedule_parents(node);
471            }
472        }
473
474        let activation_nodes = std::mem::take(&mut self.activation_nodes);
475        for node in activation_nodes {
476            if self.nodes[node.index()].needed_by > 0 && self.nodes[node.index()].stale {
477                self.schedule(node);
478            }
479        }
480
481        if self.queues.values().all(VecDeque::is_empty) {
482            self.stabilizing = false;
483        }
484    }
485
486    fn recompute(&mut self, node: ReactiveNodeIdV0) {
487        debug_assert_ne!(
488            self.nodes[node.index()].last_recomputed_wave,
489            Some(self.current_wave),
490            "a node must not recompute more than once in one wave"
491        );
492        let next = self.evaluate(node);
493        let previous = self.nodes[node.index()].state.clone();
494        let policy = self.nodes[node.index()].change_policy;
495        let changed = !policy.equivalent(&previous, &next);
496
497        self.nodes[node.index()].state = next.clone();
498        self.nodes[node.index()].stale = false;
499        self.nodes[node.index()].last_recomputed_wave = Some(self.current_wave);
500        self.nodes[node.index()].recompute_count += 1;
501
502        if !changed {
503            return;
504        }
505        if let NodeOperationV0::EffectBoundary { channel } = &self.nodes[node.index()].operation {
506            self.effect_receipts.push(EffectReceiptV0 {
507                channel: channel.clone(),
508                wave: self.current_wave,
509                state: next,
510            });
511        }
512        self.schedule_parents(node);
513    }
514
515    fn evaluate(&mut self, node: ReactiveNodeIdV0) -> ReactiveStateV0 {
516        let operation = self.nodes[node.index()].operation.clone();
517        let dependencies = self.nodes[node.index()].dependencies.clone();
518        match operation {
519            NodeOperationV0::Input | NodeOperationV0::AsyncResult => {
520                self.nodes[node.index()].state.clone()
521            }
522            NodeOperationV0::Map { operation } => {
523                operation(&self.nodes[dependencies[0].index()].state)
524            }
525            NodeOperationV0::Zip { operation } => operation(
526                &self.nodes[dependencies[0].index()].state,
527                &self.nodes[dependencies[1].index()].state,
528            ),
529            NodeOperationV0::Switch => match &self.nodes[dependencies[0].index()].state {
530                ReactiveStateV0::Available(ReactiveValueV0::Bool(false)) => {
531                    self.nodes[dependencies[1].index()].state.clone()
532                }
533                ReactiveStateV0::Available(ReactiveValueV0::Bool(true)) => {
534                    self.nodes[dependencies[2].index()].state.clone()
535                }
536                ReactiveStateV0::Available(_) => {
537                    ReactiveStateV0::Unavailable(ReactiveUnavailableV0::new(
538                        "switchSelectorTypeMismatch",
539                        "a switch selector must carry a boolean value",
540                    ))
541                }
542                ReactiveStateV0::Unavailable(unavailable) => {
543                    ReactiveStateV0::Unavailable(unavailable.clone())
544                }
545            },
546            NodeOperationV0::DeltaFold { keys } => {
547                let mut entries = std::mem::take(&mut self.nodes[node.index()].delta_entries);
548                let dirty_dependencies =
549                    std::mem::take(&mut self.nodes[node.index()].delta_dirty_dependencies);
550                let mut digest = match &self.nodes[node.index()].state {
551                    ReactiveStateV0::Available(ReactiveValueV0::Digest(digest)) => *digest,
552                    _ => [0; 32],
553                };
554                let refresh_all = entries.is_empty() || dirty_dependencies.is_empty();
555                for (key, dependency) in keys.iter().zip(dependencies) {
556                    if !refresh_all && !dirty_dependencies.contains(&dependency) {
557                        continue;
558                    }
559                    let next = self.nodes[dependency.index()].state.clone();
560                    if let Some(previous) = entries.insert(key.clone(), next.clone()) {
561                        xor_digest(&mut digest, &digest_entry(key, &previous));
562                    }
563                    xor_digest(&mut digest, &digest_entry(key, &next));
564                    self.nodes[node.index()].delta_update_count += 1;
565                }
566                self.nodes[node.index()].delta_entries = entries;
567                ReactiveStateV0::Available(ReactiveValueV0::Digest(digest))
568            }
569            NodeOperationV0::EffectBoundary { .. } => {
570                self.nodes[dependencies[0].index()].state.clone()
571            }
572        }
573    }
574
575    fn increase_necessity(&mut self, node: ReactiveNodeIdV0) {
576        self.nodes[node.index()].needed_by += 1;
577        if self.nodes[node.index()].needed_by != 1 {
578            return;
579        }
580        for dependency in self.nodes[node.index()].dependencies.clone() {
581            self.increase_necessity(dependency);
582        }
583        if !matches!(
584            self.nodes[node.index()].operation,
585            NodeOperationV0::Input | NodeOperationV0::AsyncResult
586        ) && self.nodes[node.index()].stale
587        {
588            self.activation_nodes.insert(node);
589        }
590    }
591
592    fn decrease_necessity(&mut self, node: ReactiveNodeIdV0) {
593        debug_assert!(self.nodes[node.index()].needed_by > 0);
594        self.nodes[node.index()].needed_by -= 1;
595        if self.nodes[node.index()].needed_by != 0 {
596            return;
597        }
598        for dependency in self.nodes[node.index()].dependencies.clone() {
599            self.decrease_necessity(dependency);
600        }
601    }
602
603    fn schedule_parents(&mut self, node: ReactiveNodeIdV0) {
604        for parent in self.nodes[node.index()].parents.clone() {
605            if matches!(
606                self.nodes[parent.index()].operation,
607                NodeOperationV0::DeltaFold { .. }
608            ) {
609                self.nodes[parent.index()]
610                    .delta_dirty_dependencies
611                    .insert(node);
612            }
613            if self.nodes[parent.index()].needed_by > 0 {
614                self.schedule(parent);
615            } else {
616                self.nodes[parent.index()].stale = true;
617            }
618        }
619    }
620
621    fn schedule(&mut self, node: ReactiveNodeIdV0) {
622        if self.nodes[node.index()].scheduled_wave == Some(self.current_wave) {
623            return;
624        }
625        self.nodes[node.index()].scheduled_wave = Some(self.current_wave);
626        self.queues
627            .entry(self.nodes[node.index()].height)
628            .or_default()
629            .push_back(node);
630    }
631
632    fn pop_next_scheduled(&mut self) -> Option<ReactiveNodeIdV0> {
633        loop {
634            let mut entry = self.queues.first_entry()?;
635            let queue = entry.get_mut();
636            if let Some(node) = queue.pop_front() {
637                if queue.is_empty() {
638                    entry.remove();
639                }
640                return Some(node);
641            }
642            entry.remove();
643        }
644    }
645
646    fn node(&self, node: ReactiveNodeIdV0) -> Result<&RuntimeNodeV0, ReactiveEngineErrorV0> {
647        if node.graph != self.graph_id {
648            return Err(ReactiveEngineErrorV0::ForeignNodeId {
649                node_index: node.index(),
650                expected_graph: self.graph_id.value(),
651                actual_graph: node.graph.value(),
652            });
653        }
654        self.nodes
655            .get(node.index())
656            .ok_or(ReactiveEngineErrorV0::InvalidNode {
657                node_index: node.index(),
658            })
659    }
660
661    #[cfg(test)]
662    pub(crate) fn corrupt_delta_fold_entry_for_test(
663        &mut self,
664        node: ReactiveNodeIdV0,
665        key: &str,
666        state: ReactiveStateV0,
667    ) {
668        self.nodes[node.index()]
669            .delta_entries
670            .insert(key.to_string(), state);
671    }
672
673    #[cfg(test)]
674    pub(crate) fn replace_state_without_scheduling_for_test(
675        &mut self,
676        node: ReactiveNodeIdV0,
677        state: ReactiveStateV0,
678    ) {
679        self.nodes[node.index()].state = state;
680    }
681}
682
683fn full_delta_digest(entries: &BTreeMap<String, ReactiveStateV0>) -> [u8; 32] {
684    let mut digest = [0; 32];
685    for (key, state) in entries {
686        xor_digest(&mut digest, &digest_entry(key, state));
687    }
688    digest
689}
690
691fn digest_entry(key: &str, state: &ReactiveStateV0) -> [u8; 32] {
692    let mut hasher = blake3::Hasher::new();
693    hasher.update(b"omena-reactive-delta-entry-v0");
694    hash_text(&mut hasher, key);
695    hash_state(&mut hasher, state);
696    *hasher.finalize().as_bytes()
697}
698
699fn hash_state(hasher: &mut blake3::Hasher, state: &ReactiveStateV0) {
700    match state {
701        ReactiveStateV0::Available(value) => {
702            hasher.update(&[0]);
703            hash_value(hasher, value);
704        }
705        ReactiveStateV0::Unavailable(unavailable) => {
706            hasher.update(&[1]);
707            hash_text(hasher, &unavailable.code);
708            hash_text(hasher, &unavailable.detail);
709        }
710    }
711}
712
713fn hash_value(hasher: &mut blake3::Hasher, value: &ReactiveValueV0) {
714    match value {
715        ReactiveValueV0::Unit => {
716            hasher.update(&[0]);
717        }
718        ReactiveValueV0::Bool(value) => {
719            hasher.update(&[1, u8::from(*value)]);
720        }
721        ReactiveValueV0::Counter(value) => {
722            hasher.update(&[2]);
723            hasher.update(&value.to_le_bytes());
724        }
725        ReactiveValueV0::Text(value) => {
726            hasher.update(&[3]);
727            hash_text(hasher, value);
728        }
729        ReactiveValueV0::StringSet(values) => {
730            hasher.update(&[4]);
731            hasher.update(&(values.len() as u64).to_le_bytes());
732            for value in values {
733                hash_text(hasher, value);
734            }
735        }
736        ReactiveValueV0::TextMap(values) => {
737            hasher.update(&[5]);
738            hasher.update(&(values.len() as u64).to_le_bytes());
739            for (key, value) in values {
740                hash_text(hasher, key);
741                hash_text(hasher, value);
742            }
743        }
744        ReactiveValueV0::Tuple(values) => {
745            hasher.update(&[6]);
746            hasher.update(&(values.len() as u64).to_le_bytes());
747            for value in values {
748                hash_value(hasher, value);
749            }
750        }
751        ReactiveValueV0::Digest(value) => {
752            hasher.update(&[7]);
753            hasher.update(value);
754        }
755    }
756}
757
758fn hash_text(hasher: &mut blake3::Hasher, value: &str) {
759    hasher.update(&(value.len() as u64).to_le_bytes());
760    hasher.update(value.as_bytes());
761}
762
763fn xor_digest(left: &mut [u8; 32], right: &[u8; 32]) {
764    for (left_byte, right_byte) in left.iter_mut().zip(right) {
765        *left_byte ^= right_byte;
766    }
767}
768
769#[cfg(test)]
770mod tests {
771    use std::collections::BTreeSet;
772
773    use super::*;
774    use crate::{ReactiveGraphBuilderV0, ReactiveValueV0};
775
776    fn exact_policy() -> ChangePolicyV0 {
777        ChangePolicyV0::exact("testSemanticValue")
778    }
779
780    fn counter_map(state: &ReactiveStateV0) -> ReactiveStateV0 {
781        match state {
782            ReactiveStateV0::Available(ReactiveValueV0::Counter(value)) => {
783                ReactiveStateV0::available(ReactiveValueV0::Counter(value.saturating_add(1)))
784            }
785            _ => ReactiveStateV0::unavailable("counterInputRequired", "expected a counter"),
786        }
787    }
788
789    fn counter_zip(left: &ReactiveStateV0, right: &ReactiveStateV0) -> ReactiveStateV0 {
790        match (left, right) {
791            (
792                ReactiveStateV0::Available(ReactiveValueV0::Counter(left)),
793                ReactiveStateV0::Available(ReactiveValueV0::Counter(right)),
794            ) => ReactiveStateV0::available(ReactiveValueV0::Counter(left.saturating_add(*right))),
795            _ => ReactiveStateV0::unavailable("counterInputsRequired", "expected two counters"),
796        }
797    }
798
799    fn tuple_revision(state: &ReactiveStateV0) -> ReactiveStateV0 {
800        match state {
801            ReactiveStateV0::Available(ReactiveValueV0::Tuple(values)) => values
802                .get(1)
803                .cloned()
804                .map(ReactiveStateV0::available)
805                .unwrap_or_else(|| {
806                    ReactiveStateV0::unavailable(
807                        "revisionMissing",
808                        "the tuple must carry a revision",
809                    )
810                }),
811            _ => ReactiveStateV0::unavailable("tupleRequired", "expected a tuple"),
812        }
813    }
814
815    fn always_different(_: &ReactiveStateV0, _: &ReactiveStateV0) -> bool {
816        false
817    }
818
819    fn text_field_only(previous: &ReactiveStateV0, next: &ReactiveStateV0) -> bool {
820        fn text(state: &ReactiveStateV0) -> Option<&str> {
821            let ReactiveStateV0::Available(ReactiveValueV0::Tuple(values)) = state else {
822                return None;
823            };
824            let Some(ReactiveValueV0::Text(text)) = values.first() else {
825                return None;
826            };
827            Some(text)
828        }
829        text(previous) == text(next)
830    }
831
832    fn revisioned_text(text: &str, revision: u64) -> ReactiveStateV0 {
833        ReactiveStateV0::available(ReactiveValueV0::Tuple(vec![
834            ReactiveValueV0::Text(text.to_string()),
835            ReactiveValueV0::Counter(revision),
836        ]))
837    }
838
839    fn settled(engine: &mut ReactiveEngineV0) -> Result<(), ReactiveEngineErrorV0> {
840        loop {
841            if matches!(
842                engine.stabilize_step(64)?,
843                StabilizeStatusV0::Settled { .. }
844            ) {
845                return Ok(());
846            }
847        }
848    }
849
850    #[test]
851    fn graph_exposes_all_static_node_kinds() -> Result<(), Box<dyn Error>> {
852        let mut graph = ReactiveGraphBuilderV0::new();
853        let input = graph.add_input(
854            ReactiveStateV0::available(ReactiveValueV0::Counter(1)),
855            exact_policy(),
856        );
857        let asynchronous = graph.add_async_result(
858            ReactiveStateV0::available(ReactiveValueV0::Counter(2)),
859            exact_policy(),
860        );
861        let mapped = graph.add_map(input, counter_map, exact_policy());
862        let zipped = graph.add_zip(mapped, asynchronous, counter_zip, exact_policy());
863        let selector = graph.add_input(
864            ReactiveStateV0::available(ReactiveValueV0::Bool(true)),
865            exact_policy(),
866        );
867        let switched = graph.add_switch(selector, mapped, zipped, exact_policy());
868        let folded = graph.add_delta_fold(
869            vec![
870                ("mapped".to_string(), mapped),
871                ("switched".to_string(), switched),
872            ],
873            exact_policy(),
874        )?;
875        let boundary = graph.add_effect_boundary(folded, "diagnostics", exact_policy());
876        let engine = graph.build()?;
877
878        let kinds = [
879            input,
880            mapped,
881            zipped,
882            switched,
883            folded,
884            asynchronous,
885            boundary,
886        ]
887        .into_iter()
888        .map(|node| engine.node_kind(node))
889        .collect::<Result<BTreeSet<_>, _>>()?;
890        assert_eq!(
891            kinds,
892            BTreeSet::from([
893                ReactiveNodeKindV0::Input,
894                ReactiveNodeKindV0::Map,
895                ReactiveNodeKindV0::Zip,
896                ReactiveNodeKindV0::Switch,
897                ReactiveNodeKindV0::DeltaFold,
898                ReactiveNodeKindV0::AsyncResult,
899                ReactiveNodeKindV0::EffectBoundary,
900            ])
901        );
902        Ok(())
903    }
904
905    #[test]
906    fn runtime_entry_points_reject_foreign_graph_node_ids() -> Result<(), Box<dyn Error>> {
907        let mut first_graph = ReactiveGraphBuilderV0::new();
908        let first_input = first_graph.add_input(
909            ReactiveStateV0::available(ReactiveValueV0::Counter(1)),
910            exact_policy(),
911        );
912        let first_second_input = first_graph.add_input(
913            ReactiveStateV0::available(ReactiveValueV0::Counter(2)),
914            exact_policy(),
915        );
916        let mut first_engine = first_graph.build()?;
917
918        let mut second_graph = ReactiveGraphBuilderV0::new();
919        let _second_input = second_graph.add_input(
920            ReactiveStateV0::available(ReactiveValueV0::Counter(100)),
921            exact_policy(),
922        );
923        let second_second_input = second_graph.add_input(
924            ReactiveStateV0::available(ReactiveValueV0::Counter(200)),
925            exact_policy(),
926        );
927        let _second_engine = second_graph.build()?;
928
929        for result in [
930            first_engine.node_kind(second_second_input).map(|_| ()),
931            first_engine.state(second_second_input).map(|_| ()),
932            first_engine.is_necessary(second_second_input).map(|_| ()),
933            first_engine.is_stale(second_second_input).map(|_| ()),
934            first_engine
935                .node_recompute_count(second_second_input)
936                .map(|_| ()),
937            first_engine
938                .delta_update_count(second_second_input)
939                .map(|_| ()),
940            first_engine.observe(second_second_input),
941            first_engine.unobserve(second_second_input),
942            first_engine.deposit(
943                second_second_input,
944                ReactiveStateV0::available(ReactiveValueV0::Counter(777)),
945            ),
946        ] {
947            assert!(matches!(
948                result,
949                Err(ReactiveEngineErrorV0::ForeignNodeId {
950                    node_index: 1,
951                    expected_graph,
952                    actual_graph,
953                }) if expected_graph != actual_graph
954            ));
955        }
956        assert!(matches!(
957            first_engine.verify_delta_fold(second_second_input),
958            Err(DeltaFoldParityErrorV0::ForeignNodeId {
959                node_index: 1,
960                expected_graph,
961                actual_graph,
962            }) if expected_graph != actual_graph
963        ));
964        assert_eq!(
965            first_engine.state(first_second_input)?,
966            &ReactiveStateV0::available(ReactiveValueV0::Counter(2))
967        );
968        assert_eq!(
969            first_engine.state(first_input)?,
970            &ReactiveStateV0::available(ReactiveValueV0::Counter(1))
971        );
972        Ok(())
973    }
974
975    #[test]
976    fn deposits_arriving_mid_wave_wait_for_the_next_wave() -> Result<(), Box<dyn Error>> {
977        let mut graph = ReactiveGraphBuilderV0::new();
978        let input = graph.add_input(
979            ReactiveStateV0::available(ReactiveValueV0::Counter(0)),
980            exact_policy(),
981        );
982        let mapped = graph.add_map(input, counter_map, exact_policy());
983        let boundary = graph.add_effect_boundary(mapped, "result", exact_policy());
984        let mut engine = graph.build()?;
985        engine.observe(boundary)?;
986        engine.deposit(
987            input,
988            ReactiveStateV0::available(ReactiveValueV0::Counter(1)),
989        )?;
990
991        assert!(matches!(
992            engine.stabilize_step(1)?,
993            StabilizeStatusV0::Pending { .. }
994        ));
995        engine.deposit(
996            input,
997            ReactiveStateV0::available(ReactiveValueV0::Counter(2)),
998        )?;
999        settled(&mut engine)?;
1000        assert_eq!(
1001            engine.drain_effect_receipts(),
1002            vec![EffectReceiptV0 {
1003                channel: "result".to_string(),
1004                wave: 1,
1005                state: ReactiveStateV0::available(ReactiveValueV0::Counter(2)),
1006            }]
1007        );
1008
1009        settled(&mut engine)?;
1010        assert_eq!(
1011            engine.drain_effect_receipts(),
1012            vec![EffectReceiptV0 {
1013                channel: "result".to_string(),
1014                wave: 2,
1015                state: ReactiveStateV0::available(ReactiveValueV0::Counter(3)),
1016            }]
1017        );
1018        Ok(())
1019    }
1020
1021    #[test]
1022    fn delta_fold_is_checked_against_a_full_rebuild() -> Result<(), Box<dyn Error>> {
1023        let mut graph = ReactiveGraphBuilderV0::new();
1024        let first = graph.add_input(
1025            ReactiveStateV0::available(ReactiveValueV0::Counter(1)),
1026            exact_policy(),
1027        );
1028        let second = graph.add_input(
1029            ReactiveStateV0::available(ReactiveValueV0::Counter(2)),
1030            exact_policy(),
1031        );
1032        let fold = graph.add_delta_fold(
1033            vec![("first".to_string(), first), ("second".to_string(), second)],
1034            exact_policy(),
1035        )?;
1036        let mut engine = graph.build()?;
1037        engine.observe(fold)?;
1038        settled(&mut engine)?;
1039        engine.verify_delta_fold(fold)?;
1040        assert_eq!(engine.delta_update_count(fold)?, 2);
1041
1042        engine.deposit(
1043            first,
1044            ReactiveStateV0::available(ReactiveValueV0::Counter(3)),
1045        )?;
1046        settled(&mut engine)?;
1047        engine.verify_delta_fold(fold)?;
1048        assert_eq!(
1049            engine.delta_update_count(fold)?,
1050            3,
1051            "only the changed key should be revisited after the initial rebuild"
1052        );
1053
1054        engine.corrupt_delta_fold_entry_for_test(
1055            fold,
1056            "first",
1057            ReactiveStateV0::available(ReactiveValueV0::Counter(99)),
1058        );
1059        assert!(matches!(
1060            engine.verify_delta_fold(fold),
1061            Err(DeltaFoldParityErrorV0::Diverged {
1062                incremental_digest,
1063                rebuilt_digest,
1064                ..
1065            }) if incremental_digest == rebuilt_digest
1066        ));
1067        Ok(())
1068    }
1069
1070    #[test]
1071    fn delta_fold_detects_dependency_changes_missed_by_dirty_tracking() -> Result<(), Box<dyn Error>>
1072    {
1073        let mut graph = ReactiveGraphBuilderV0::new();
1074        let first = graph.add_input(
1075            ReactiveStateV0::available(ReactiveValueV0::Counter(1)),
1076            exact_policy(),
1077        );
1078        let second = graph.add_input(
1079            ReactiveStateV0::available(ReactiveValueV0::Counter(2)),
1080            exact_policy(),
1081        );
1082        let fold = graph.add_delta_fold(
1083            vec![("first".to_string(), first), ("second".to_string(), second)],
1084            exact_policy(),
1085        )?;
1086        let mut engine = graph.build()?;
1087        engine.observe(fold)?;
1088        settled(&mut engine)?;
1089
1090        engine.replace_state_without_scheduling_for_test(
1091            first,
1092            ReactiveStateV0::available(ReactiveValueV0::Counter(999)),
1093        );
1094
1095        assert!(matches!(
1096            engine.verify_delta_fold(fold),
1097            Err(DeltaFoldParityErrorV0::Diverged {
1098                incremental_digest,
1099                rebuilt_digest,
1100                ..
1101            }) if incremental_digest != rebuilt_digest
1102        ));
1103        Ok(())
1104    }
1105
1106    #[test]
1107    fn dependency_rebuild_does_not_read_incremental_entries() {
1108        let source = include_str!("engine.rs");
1109        let marker = "fn rebuild_delta_fold_from_dependencies";
1110        let start = source.find(marker).unwrap_or(source.len());
1111        assert!(
1112            start < source.len(),
1113            "dependency rebuild function is missing"
1114        );
1115        let after_start = &source[start..];
1116        let end = after_start
1117            .find("\n    fn begin_wave")
1118            .unwrap_or(after_start.len());
1119        assert!(
1120            end < after_start.len(),
1121            "dependency rebuild function boundary is missing"
1122        );
1123        let function = &after_start[..end];
1124        assert!(
1125            !function.contains("delta_entries"),
1126            "dependency rebuild must not read incremental fold entries"
1127        );
1128    }
1129
1130    #[test]
1131    fn unavailable_nodes_do_not_poison_independent_branches() -> Result<(), Box<dyn Error>> {
1132        let mut graph = ReactiveGraphBuilderV0::new();
1133        let unavailable = graph.add_async_result(
1134            ReactiveStateV0::unavailable("queryCancelled", "the query was cancelled"),
1135            exact_policy(),
1136        );
1137        let healthy = graph.add_input(
1138            ReactiveStateV0::available(ReactiveValueV0::Counter(7)),
1139            exact_policy(),
1140        );
1141        let failed_map = graph.add_map(unavailable, counter_map, exact_policy());
1142        let healthy_map = graph.add_map(healthy, counter_map, exact_policy());
1143        let mut engine = graph.build()?;
1144        engine.observe(failed_map)?;
1145        engine.observe(healthy_map)?;
1146        settled(&mut engine)?;
1147
1148        assert!(matches!(
1149            engine.state(failed_map)?,
1150            ReactiveStateV0::Unavailable(_)
1151        ));
1152        assert_eq!(
1153            engine.state(healthy_map)?,
1154            &ReactiveStateV0::available(ReactiveValueV0::Counter(8))
1155        );
1156        Ok(())
1157    }
1158
1159    #[test]
1160    fn unobserved_nodes_stay_stale_until_needed() -> Result<(), Box<dyn Error>> {
1161        let mut graph = ReactiveGraphBuilderV0::new();
1162        let input = graph.add_input(
1163            ReactiveStateV0::available(ReactiveValueV0::Counter(1)),
1164            exact_policy(),
1165        );
1166        let mapped = graph.add_map(input, counter_map, exact_policy());
1167        let mut engine = graph.build()?;
1168        engine.deposit(
1169            input,
1170            ReactiveStateV0::available(ReactiveValueV0::Counter(2)),
1171        )?;
1172        settled(&mut engine)?;
1173        assert!(engine.is_stale(mapped)?);
1174        assert_eq!(engine.node_recompute_count(mapped)?, 0);
1175
1176        engine.observe(mapped)?;
1177        settled(&mut engine)?;
1178        assert!(!engine.is_stale(mapped)?);
1179        assert_eq!(engine.node_recompute_count(mapped)?, 1);
1180        Ok(())
1181    }
1182
1183    #[test]
1184    fn bounded_steps_recompute_each_node_at_most_once_per_wave() -> Result<(), Box<dyn Error>> {
1185        let mut graph = ReactiveGraphBuilderV0::new();
1186        let input = graph.add_input(
1187            ReactiveStateV0::available(ReactiveValueV0::Counter(0)),
1188            exact_policy(),
1189        );
1190        let first = graph.add_map(input, counter_map, exact_policy());
1191        let second = graph.add_map(first, counter_map, exact_policy());
1192        let boundary = graph.add_effect_boundary(second, "result", exact_policy());
1193        let mut engine = graph.build()?;
1194        engine.observe(boundary)?;
1195        engine.deposit(
1196            input,
1197            ReactiveStateV0::available(ReactiveValueV0::Counter(1)),
1198        )?;
1199
1200        let mut steps = 0;
1201        loop {
1202            steps += 1;
1203            if matches!(engine.stabilize_step(1)?, StabilizeStatusV0::Settled { .. }) {
1204                break;
1205            }
1206        }
1207        assert_eq!(steps, 3);
1208        assert_eq!(engine.node_recompute_count(first)?, 1);
1209        assert_eq!(engine.node_recompute_count(second)?, 1);
1210        assert_eq!(engine.node_recompute_count(boundary)?, 1);
1211        Ok(())
1212    }
1213
1214    #[test]
1215    fn semantic_policy_avoids_spurious_refires_for_equal_allocations() -> Result<(), Box<dyn Error>>
1216    {
1217        let mut exact_graph = ReactiveGraphBuilderV0::new();
1218        let exact_input = exact_graph.add_input(revisioned_text("same", 1), exact_policy());
1219        let exact_map = exact_graph.add_map(exact_input, tuple_revision, exact_policy());
1220        let mut exact_engine = exact_graph.build()?;
1221        exact_engine.observe(exact_map)?;
1222        settled(&mut exact_engine)?;
1223        exact_engine.deposit(exact_input, revisioned_text("same", 1))?;
1224        settled(&mut exact_engine)?;
1225        assert_eq!(exact_engine.node_recompute_count(exact_map)?, 1);
1226
1227        let mut identity_graph = ReactiveGraphBuilderV0::new();
1228        let identity_input = identity_graph.add_input(
1229            revisioned_text("same", 1),
1230            ChangePolicyV0::custom("allocationIdentity", always_different),
1231        );
1232        let identity_map = identity_graph.add_map(identity_input, tuple_revision, exact_policy());
1233        let mut identity_engine = identity_graph.build()?;
1234        identity_engine.observe(identity_map)?;
1235        settled(&mut identity_engine)?;
1236        identity_engine.deposit(identity_input, revisioned_text("same", 1))?;
1237        settled(&mut identity_engine)?;
1238        assert_eq!(identity_engine.node_recompute_count(identity_map)?, 2);
1239        Ok(())
1240    }
1241
1242    #[test]
1243    fn semantic_policy_preserves_updates_hidden_by_partial_equality() -> Result<(), Box<dyn Error>>
1244    {
1245        let mut exact_graph = ReactiveGraphBuilderV0::new();
1246        let exact_input = exact_graph.add_input(revisioned_text("stable", 1), exact_policy());
1247        let exact_map = exact_graph.add_map(exact_input, tuple_revision, exact_policy());
1248        let mut exact_engine = exact_graph.build()?;
1249        exact_engine.observe(exact_map)?;
1250        settled(&mut exact_engine)?;
1251        exact_engine.deposit(exact_input, revisioned_text("stable", 2))?;
1252        settled(&mut exact_engine)?;
1253        assert_eq!(
1254            exact_engine.state(exact_map)?,
1255            &ReactiveStateV0::available(ReactiveValueV0::Counter(2))
1256        );
1257
1258        let mut partial_graph = ReactiveGraphBuilderV0::new();
1259        let partial_input = partial_graph.add_input(
1260            revisioned_text("stable", 1),
1261            ChangePolicyV0::custom("textFieldOnly", text_field_only),
1262        );
1263        let partial_map = partial_graph.add_map(partial_input, tuple_revision, exact_policy());
1264        let mut partial_engine = partial_graph.build()?;
1265        partial_engine.observe(partial_map)?;
1266        settled(&mut partial_engine)?;
1267        partial_engine.deposit(partial_input, revisioned_text("stable", 2))?;
1268        settled(&mut partial_engine)?;
1269        assert_eq!(
1270            partial_engine.state(partial_map)?,
1271            &ReactiveStateV0::available(ReactiveValueV0::Counter(1))
1272        );
1273        Ok(())
1274    }
1275}
omena_reactive/engine.rs

omena_reactive/
engine.rs
