async_nodes/

merge.rs

//! Fork/join branch merging.
//!
//! BSP supersteps run frontier nodes concurrently, so each frontier task executes against
//! its own clone of the consumer state (Rust's aliasing rules make a shared `&mut S` into
//! N concurrent futures impossible — the same reason the reference consumer passes state
//! by value per branch). When the superstep barrier folds the branch states back into the
//! trunk, a [`Merger`] decides *how*:
//!
//! - A join node names a merger (consumer-registered, like named conditions), and the
//!   engine folds branch states into the trunk **in frontier declaration order** — never
//!   in task completion order, which would make runs irreproducible.
//! - Without a named merger the fold is last-writer-wins (still declaration order, still
//!   deterministic), and validation emits a warning — silent last-wins is how subtle state
//!   loss hides, and the warning is the point.
//!
//! Because every branch clones the trunk *at its superstep start*, writes folded at earlier
//! barriers are visible to branches that start later; only same-superstep siblings are
//! isolated from each other, and the merger resolves exactly those sibling conflicts.

use std::{collections::HashMap, sync::Arc};

use async_trait::async_trait;

use super::error::BspError;

/// How a branched state rejoins the trunk.
///
/// Implementations are consumer-owned: the engine knows nothing about what `S` fields mean.
/// A typical merger replays a branch's delta into the trunk (e.g. folding a child store's
/// integer deltas and rolling up metrics, the way the reference consumer's parallel tool
/// map-reduce does).
#[async_trait]
pub trait Merger<S>: Send + Sync {
    /// Folds one finished branch state back into the trunk.
    fn merge(&self, trunk: &mut S, branch: S);

    /// Finalizes a barrier fold: called exactly once after the last `merge` of a
    /// multi-node superstep barrier (never called for single-node frontiers, which
    /// mutate the trunk directly and never fold).
    ///
    /// This is the hook for *all-at-once* merge semantics: `merge` can accumulate the
    /// branch states (interior mutability) and `finish` collapses them into the trunk in
    /// one shot — e.g. the reference consumer's reducer barriers fold a whole
    /// `Vec<GraphState>` through one async, fallible reduction instead of pairwise folds.
    ///
    /// It is `async` because barrier finalization may itself be a computation (the
    /// reference consumer's reducers call LLMs), and fallible so a failed collapse
    /// aborts the run loudly instead of silently continuing with an unreduced trunk.
    /// The default is a no-op `Ok(())`, which is what pairwise `merge`-only mergers want.
    async fn finish(&self, _trunk: &mut S) -> Result<(), BspError> {
        Ok(())
    }
}

/// Any `Fn(&mut S, S)` closure is a merger.
#[async_trait]
impl<S, F> Merger<S> for F
where
    F: Fn(&mut S, S) + Send + Sync,
{
    fn merge(&self, trunk: &mut S, branch: S) {
        self(trunk, branch);
    }
}

/// Map of merger name -> merger, carried by the graph.
pub type MergerMap<S> = HashMap<String, Arc<dyn Merger<S>>>;

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_closure_is_a_merger() {
        let merger = |trunk: &mut Vec<String>, branch: Vec<String>| {
            for item in branch {
                if !trunk.contains(&item) {
                    trunk.push(item);
                }
            }
        };
        let mut trunk = vec!["a".to_string()];
        Merger::merge(&merger, &mut trunk, vec!["a".to_string(), "b".to_string()]);
        assert_eq!(trunk, vec!["a".to_string(), "b".to_string()]);
    }

    #[test]
    fn test_fold_declaration_order_is_deterministic() {
        // Simulates the engine's barrier fold: branches folded in declaration order.
        let append = |trunk: &mut Vec<&'static str>, mut branch: Vec<&'static str>| {
            trunk.append(&mut branch);
        };
        let branches = [vec!["a"], vec!["b"], vec!["c"]];
        let mut trunk = Vec::new();
        for branch in branches {
            Merger::merge(&append, &mut trunk, branch);
        }
        assert_eq!(trunk, vec!["a", "b", "c"]);
    }
}