snix_castore/directoryservice/

directory_graph.rs

use petgraph::{
    graph::{DiGraph, NodeIndex},
    visit::{Bfs, DfsPostOrder, Walker},
};
use std::collections::{HashMap, HashSet, hash_map};
use tracing::{instrument, warn};

use crate::directoryservice::order_validator::OrderingError;
use crate::{B3Digest, Directory, Node};

#[derive(PartialEq, Eq, Debug)]
pub enum DirectoryOrder {
    RootToLeaves,
    LeavesToRoot,
}

/// This represents a full (and validated) graph of [Directory] nodes.
/// It can be constructed using [DirectoryGraphBuilder], and is normally used to
/// convert from one order to the other.
/// If you just want to validate an order without keeping the results,
/// `RootToLeavesValidator` or `LeavesToRootValidator` can be used.
#[derive(Default)]
pub struct DirectoryGraph {
    // A directed graph, using Directory as node weight.
    // Edges point from parents to children.
    graph: DiGraph<Directory, ()>,

    // Points to the root.
    root_idx: NodeIndex,
}

impl DirectoryGraph {
    /// Drains the graph, returning node weights in the chosen [DirectoryOrder].
    #[instrument(level = "trace", skip_all)]
    pub fn drain(self, order: DirectoryOrder) -> impl Iterator<Item = Directory> {
        let order = match order {
            DirectoryOrder::RootToLeaves => {
                // do a BFS traversal of the graph, starting with the root node
                Bfs::new(&self.graph, self.root_idx)
                    .iter(&self.graph)
                    .collect::<Vec<_>>()
            }
            DirectoryOrder::LeavesToRoot => {
                // do a DFS Post-Order traversal of the graph, starting with the root node
                DfsPostOrder::new(&self.graph, self.root_idx)
                    .iter(&self.graph)
                    .collect::<Vec<_>>()
            }
        };

        let (mut nodes, _edges) = self.graph.into_nodes_edges();
        order
            .into_iter()
            .map(move |i| std::mem::take(&mut nodes[i.index()].weight))
    }

    pub fn root(&self) -> &Directory {
        self.graph
            .node_weight(self.root_idx)
            .expect("Snix bug: root not found")
    }
}

/// This allows constructing a [DirectoryGraph].
/// After deciding on the insertion order (via [Self::new_with_insertion_order]),
/// different [Directory] can be passed to [Self::insert].
/// A [Self::build] consumes the builder, returning a validated [DirectoryGraph],
/// or an error.
/// The resulting [DirectoryGraph] can be used to drain the graph in either order.
///
/// It does do the same checks as `RootToLeavesValidator` and `LeavesToRootValidator`
/// (insertion order, completeness, connectivity, correct sizes referenced).
// NOTE: a child is always smaller than its parent
pub struct DirectoryGraphBuilder {
    /// The order of [Directory] elements [Self::insert] is called with.
    insertion_order: DirectoryOrder,

    /// A directed graph, using Directory as node weight.
    /// Edges point from parents to children.
    graph: DiGraph<Directory, ()>,

    /// A lookup table from directory digest to node index and size.
    /// The size is stored to avoid having to calculate it multiple times.
    digest_to_node_idx_size: HashMap<B3Digest, (NodeIndex, u64)>,

    /// A map from digest to size and all node indexes that are pointing to it.
    /// Used in the RTL case for all unfinished edges.
    rtl_edges_todo: HashMap<B3Digest, (u64, Vec<NodeIndex>)>,

    /// Points to the first node index.
    /// Populated in the RTL case only.
    first_idx: Option<NodeIndex>,
}

impl DirectoryGraphBuilder {
    pub fn new_with_insertion_order(insertion_order: DirectoryOrder) -> Self {
        Self {
            insertion_order,
            graph: Default::default(),
            digest_to_node_idx_size: Default::default(),
            rtl_edges_todo: Default::default(),
            first_idx: None,
        }
    }

    #[instrument(level = "trace", skip_all, fields(directory.digest=%directory.digest()))]
    pub fn insert(&mut self, directory: Directory) -> Result<(), OrderingError> {
        let directory_digest = directory.digest();
        let directory_size = directory.size();

        let entry = match self
            .digest_to_node_idx_size
            .entry(directory_digest.to_owned())
        {
            hash_map::Entry::Occupied(_) => {
                warn!("directory received multiple times");
                return Ok(());
            }
            hash_map::Entry::Vacant(vacant_entry) => vacant_entry,
        };

        let node_idx = self.graph.add_node(directory);
        entry.insert((node_idx, directory_size));

        if self.insertion_order == DirectoryOrder::RootToLeaves {
            // If this was the first inserted node, set first_idx.
            // We also obviously won't find ourselves in [self.rtl_edges_todo],
            // as we're the first element.
            if self.graph.node_count() == 1 {
                self.first_idx = Some(node_idx);
            } else if let Some((digest, (size, src_idxs))) =
                // Check for our own digest in [self.rtl_edges_todo], pop and add edges to graph
                self.rtl_edges_todo.remove_entry(&directory_digest)
            {
                if size != directory_size {
                    Err(OrderingError::WrongSize { digest, size })?
                }

                for src_idx in src_idxs {
                    self.graph.add_edge(src_idx, node_idx, ());
                }
            } else {
                let directory = self
                    .graph
                    .node_weight(node_idx)
                    .expect("Snix bug: node not found")
                    .to_owned();

                Err(OrderingError::Unexpected { directory })?
            }
        }

        // Look at outgoing digests. For this we have to retrieve the previously-inserted Directory again.
        // We copy out the digests (as all code paths add edges, which mutates the graph).
        let directory = self
            .graph
            .node_weight(node_idx)
            .expect("Snix bug: node not found");
        let out_digests_sizes = directory
            .nodes()
            .filter_map(|(_, node)| {
                if let Node::Directory { digest, size } = node {
                    Some((digest.to_owned(), *size))
                } else {
                    None
                }
            })
            .collect::<Vec<_>>();

        for (out_digest, out_size) in out_digests_sizes {
            match self.insertion_order {
                DirectoryOrder::RootToLeaves => {
                    // Add outgoing pointers to the graph, or to [self.rtl_edges_todo], if not yet known.
                    if let Some((out_node_idx, seen_dir_size)) =
                        self.digest_to_node_idx_size.get(&out_digest)
                    {
                        // check size
                        if *seen_dir_size != out_size {
                            Err(OrderingError::WrongSize {
                                digest: out_digest,
                                size: out_size,
                            })?
                        }

                        // draw edge
                        self.graph.add_edge(node_idx, *out_node_idx, ());
                    } else {
                        // pointer points to something not yet in the graph, add to todo
                        match self.rtl_edges_todo.entry(out_digest) {
                            hash_map::Entry::Occupied(mut occupied_entry) => {
                                let size = occupied_entry.get().0;
                                if size != out_size {
                                    Err(OrderingError::WrongSize {
                                        digest: occupied_entry.key().to_owned(),
                                        size,
                                    })?
                                }
                                occupied_entry.get_mut().1.push(node_idx);
                            }
                            hash_map::Entry::Vacant(vacant_entry) => {
                                vacant_entry.insert((out_size, vec![node_idx]));
                            }
                        }
                    }
                }
                DirectoryOrder::LeavesToRoot => {
                    // Check all pointers in the currently added directory have already been added previously;
                    // each sent directory may only refer to directories already sent.
                    if let Some((out_node_idx, seen_dir_size)) =
                        self.digest_to_node_idx_size.get(&out_digest)
                    {
                        // check the size from the pointer matches actual size
                        if *seen_dir_size != out_size {
                            Err(OrderingError::WrongSize {
                                digest: out_digest,
                                size: out_size,
                            })?
                        }

                        // draw the edge
                        self.graph.add_edge(node_idx, *out_node_idx, ());
                    } else {
                        let directory = self
                            .graph
                            .node_weight(node_idx)
                            .expect("Snix bug: node not found");

                        Err(OrderingError::UnknownLTR {
                            digest: out_digest.clone(),
                            parent_digest: directory_digest.to_owned(),
                            path_component: directory
                                .nodes()
                                .find_map(|(path_component, node)| {
                                    if let Node::Directory { digest, .. } = node
                                        && digest == &out_digest
                                    {
                                        Some(path_component)
                                    } else {
                                        None
                                    }
                                })
                                .expect("PathComponent not found")
                                .to_owned(),
                        })?
                    }
                }
            }
        }

        Ok(())
    }

    pub fn build(self) -> Result<DirectoryGraph, OrderingError> {
        match self.insertion_order {
            // We must have received the root, and there may not be any rtl_edges_todo.
            DirectoryOrder::RootToLeaves => match self.first_idx {
                None => Err(OrderingError::NoNodesReceived),
                Some(root_idx) => {
                    if !self.rtl_edges_todo.is_empty() {
                        return Err(OrderingError::MoreThanOnePending(HashSet::from_iter(
                            self.rtl_edges_todo.into_keys(),
                        )));
                    }

                    debug_assert_eq!(
                        self.graph.externals(petgraph::Incoming).count(),
                        1,
                        "one incoming"
                    );
                    Ok(DirectoryGraph {
                        graph: self.graph,
                        root_idx,
                    })
                }
            },
            DirectoryOrder::LeavesToRoot => {
                let incomings = self.graph.externals(petgraph::Incoming).collect::<Vec<_>>();

                if incomings.is_empty() {
                    return Err(OrderingError::NoNodesReceived);
                }

                if incomings.len() != 1 {
                    return Err(OrderingError::MoreThanOnePending(HashSet::from_iter(
                        incomings.iter().map(|i| {
                            self.graph
                                .node_weight(*i)
                                .expect("Snix bug: node not found")
                                .digest()
                        }),
                    )));
                }
                Ok(DirectoryGraph {
                    graph: self.graph,
                    root_idx: incomings[0],
                })
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::directoryservice::DirectoryOrder;
    use crate::directoryservice::directory_graph::DirectoryGraphBuilder;
    use crate::fixtures::{DIRECTORY_A, DIRECTORY_B, DIRECTORY_C};
    use crate::{Directory, Node};
    use rstest::rstest;
    use std::sync::LazyLock;

    pub static BROKEN_PARENT_DIRECTORY: LazyLock<Directory> = LazyLock::new(|| {
        Directory::try_from_iter([(
            "foo".try_into().unwrap(),
            Node::Directory {
                digest: DIRECTORY_A.digest(),
                size: DIRECTORY_A.size() + 42, // wrong!
            },
        )])
        .unwrap()
    });

    #[rstest]
    /// Uploading no directories at all should fail, the empty graph is invalid.
    #[case::ltr_empty_graph(DirectoryOrder::LeavesToRoot, &[], false, None)]
    /// Uploading an empty directory should succeed.
    #[case::ltr_empty_directory(DirectoryOrder::LeavesToRoot, &[&*DIRECTORY_A], false, Some(vec![&*DIRECTORY_A]))]
    /// Uploading A, then B (referring to A) should succeed.
    #[case::ltr_simple_closure(DirectoryOrder::LeavesToRoot, &[&*DIRECTORY_A, &*DIRECTORY_B], false, Some(vec![&*DIRECTORY_A, &*DIRECTORY_B]))]
    /// Uploading A, then A, then C (referring to A twice) should succeed.
    /// We pretend to be a dumb client not deduping directories.
    #[case::ltr_same_child(DirectoryOrder::LeavesToRoot, &[&*DIRECTORY_A, &*DIRECTORY_A, &*DIRECTORY_C], false, Some(vec![&*DIRECTORY_A, &*DIRECTORY_C]))]
    /// Uploading A, then C (referring to A twice) should succeed.
    #[case::ltr_same_child_dedup(DirectoryOrder::LeavesToRoot, &[&*DIRECTORY_A, &*DIRECTORY_C], false, Some(vec![&*DIRECTORY_A, &*DIRECTORY_C]))]
    /// Uploading A, then C (referring to A twice), then B (itself referring to A) should fail during close,
    /// as B itself would be left unconnected.
    #[case::ltr_unconnected_node(DirectoryOrder::LeavesToRoot, &[&*DIRECTORY_A, &*DIRECTORY_C, &*DIRECTORY_B], false, None)]
    /// Uploading B (referring to A) should fail immediately, because A was never uploaded.
    #[case::ltr_dangling_pointer(DirectoryOrder::LeavesToRoot, &[&*DIRECTORY_B], true, None)]
    /// Uploading a directory which refers to another Directory with a wrong size should fail.
    #[case::ltr_wrong_size_in_parent(DirectoryOrder::LeavesToRoot, &[&*DIRECTORY_A, &*BROKEN_PARENT_DIRECTORY], true, None)]
    /// Downloading an empty directory should succeed.
    #[case::rtl_empty_directory(DirectoryOrder::RootToLeaves, &[&*DIRECTORY_A], false, Some(vec![&*DIRECTORY_A]))]
    /// Downlading B, then A (referenced by B) should succeed.
    #[case::rtl_simple_closure(DirectoryOrder::RootToLeaves, &[&*DIRECTORY_B, &*DIRECTORY_A], false, Some(vec![&*DIRECTORY_A, &*DIRECTORY_B]))]
    /// Downloading C (referring to A twice), then A should succeed.
    #[case::rtl_same_child_dedup(DirectoryOrder::RootToLeaves, &[&*DIRECTORY_C, &*DIRECTORY_A], false, Some(vec![&*DIRECTORY_A, &*DIRECTORY_C]))]
    /// Downloading C, then B (both referring to A but not referring to each other) should fail immediately as B has no connection to C (the root)
    #[case::rtl_unconnected_node(DirectoryOrder::RootToLeaves, &[&*DIRECTORY_C, &*DIRECTORY_B], true, None)]
    /// Downloading a directory which refers to another Directory with a wrong size should fail.
    #[case::rtl_wrong_size_in_parent(DirectoryOrder::RootToLeaves, &[&*BROKEN_PARENT_DIRECTORY, &*DIRECTORY_A], true, None)]
    fn directory_graph(
        #[case] insertion_order: DirectoryOrder,
        #[case] directories_to_upload: &[&Directory],
        #[case] exp_fail_upload_last: bool,
        #[case] exp_build: Option<Vec<&Directory>>, // Some(_) if finalize successful, None if not.
    ) {
        let mut builder = DirectoryGraphBuilder::new_with_insertion_order(insertion_order);
        let mut it = directories_to_upload.iter().peekable();

        while let Some(d) = it.next() {
            if it.peek().is_none() /* is last */ && exp_fail_upload_last {
                builder
                    .insert((*d).to_owned())
                    .expect_err("last insert to fail");
            } else {
                builder.insert((*d).to_owned()).expect("insert to succeed");
            }
        }

        if exp_fail_upload_last {
            return;
        }

        if let Some(exp_drain_ltr) = exp_build {
            let directory_graph = builder.build().expect("build to succeed");

            // drain
            let drained_ltr = directory_graph
                .drain(super::DirectoryOrder::LeavesToRoot)
                .collect::<Vec<_>>();

            assert_eq!(
                exp_drain_ltr
                    .iter()
                    .map(|d| (*d).to_owned())
                    .collect::<Vec<_>>(),
                drained_ltr
            );
        } else {
            assert!(builder.build().is_err(), "expected build to fail");
        }
    }
}