nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

// SPDX-License-Identifier: BUSL-1.1

//! `cross_core_bfs` — multi-hop BFS that drives the tree DDL aggregates
//! (`TREE_SUM`, `TREE_CHILDREN`) and any other breadth-first walk that
//! needs a flat reachable-node set across the full cross-core /
//! cross-shard neighborhood of each frontier node.
//!
//! The shared per-hop scatter/decode/merge logic lives in
//! [`super::hop::execute_neighbor_hop`]; this dispatcher only retains
//! the merged destination set. `GRAPH TRAVERSE`, which needs the
//! `{nodes,edges}` subgraph shape the remote client decodes, lives in
//! [`super::traverse_subgraph::cross_core_traverse_subgraph`].

use std::collections::HashSet;

use crate::bridge::envelope::Response;
use crate::control::state::SharedState;
use crate::engine::graph::traversal_options::GraphTraversalOptions;
use crate::types::{DatabaseId, TenantId};

use super::helpers::{encode_path, ok_response};
use super::hop::{NeighborHopParams, execute_neighbor_hop};

/// Parameters for [`cross_core_bfs_with_options`].
pub struct CrossCoreBfsParams<'a> {
    pub tenant_id: TenantId,
    pub database_id: DatabaseId,
    pub start_nodes: Vec<String>,
    pub edge_label: Option<String>,
    pub direction: crate::engine::graph::edge_store::Direction,
    pub max_depth: usize,
    pub options: &'a GraphTraversalOptions,
}

/// Cross-core BFS with explicit traversal options (fan-out limits, partial mode).
///
/// This is the cluster-aware entry point. Callers pass
/// `&GraphTraversalOptions::default()` for standard traversal.
pub async fn cross_core_bfs_with_options(
    shared: &SharedState,
    params: CrossCoreBfsParams<'_>,
) -> crate::Result<Response> {
    let CrossCoreBfsParams {
        tenant_id,
        database_id,
        start_nodes,
        edge_label,
        direction,
        max_depth,
        options,
    } = params;
    let mut visited: HashSet<String> = HashSet::new();
    let mut all_discovered: Vec<String> = Vec::new();
    let mut frontier: Vec<String> = start_nodes;

    for node in &frontier {
        visited.insert(node.clone());
        all_discovered.push(node.clone());
    }

    for _depth in 0..max_depth {
        if frontier.is_empty() {
            break;
        }

        let hop = execute_neighbor_hop(
            shared,
            tenant_id,
            database_id,
            NeighborHopParams {
                frontier: &frontier,
                edge_label: edge_label.as_deref(),
                direction,
                options,
                discovered_so_far: all_discovered.len(),
            },
        )
        .await?;

        // Extend global visited set and compute next frontier.
        let mut next_frontier: Vec<String> = Vec::new();
        for node in hop.merged_destinations {
            if visited.insert(node.clone()) {
                next_frontier.push(node.clone());
                all_discovered.push(node);
                if all_discovered.len() >= options.max_visited {
                    break;
                }
            }
        }

        frontier = next_frontier;

        if all_discovered.len() >= options.max_visited {
            break;
        }
    }

    Ok(ok_response(encode_path(&all_discovered)?))
}