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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157

// SPDX-License-Identifier: BUSL-1.1

//! Shared cluster-routing helpers for graph scatter paths (`match_scatter` and
//! `bsp_pagerank`): resolve a vShard to a live `RouteDecision`, and fetch the
//! gateway `Arc<SharedState>` used for remote dispatch.
//!
//! Both helpers resolve against LIVE Raft leadership where available so a stale
//! routing-table hint cannot misdirect a scatter. Factored here so the MATCH
//! scatter and the BSP PageRank coordinator share one implementation instead of
//! duplicating the routing-lock + live-leader plumbing.

use std::sync::Arc;

use crate::bridge::envelope::{Payload, PhysicalPlan};
use crate::control::gateway::dispatcher::{DispatchRouteParams, dispatch_route};
use crate::control::gateway::router::resolve_decision;
use crate::control::gateway::version_set::GatewayVersionSet;
use crate::control::gateway::{RouteDecision, TaskRoute};
use crate::control::server::exchange::execute_plan_all_local_cores;
use crate::control::state::SharedState;
use crate::types::{DatabaseId, TenantId, TraceId};

/// Resolve a vShard to a `RouteDecision` against live Raft leadership, falling
/// back to the routing-table hint when no live snapshot is available.
///
/// `pub(crate)` so the in-transaction staging choke points
/// (`session::leader_forward`) can resolve a staged write's / overlay drop's
/// target leader with the same live-leader semantics the graph scatter uses,
/// instead of duplicating the routing-lock + live-leader plumbing.
pub(crate) fn resolve_for_vshard(state: &SharedState, vshard_id: u32) -> RouteDecision {
    let routing_guard = state
        .cluster_routing
        .as_ref()
        .map(|rw| rw.read().unwrap_or_else(|p| p.into_inner()));
    let raft_snapshot: Vec<nodedb_cluster::GroupStatus> =
        state.raft_status_fn.get().map(|f| f()).unwrap_or_default();
    let live_leader = move |group_id: u64| -> u64 {
        raft_snapshot
            .iter()
            .find(|gs| gs.group_id == group_id)
            .map(|gs| gs.leader_id)
            .unwrap_or(0)
    };
    let live_lookup: Option<&dyn Fn(u64) -> u64> = if state.raft_status_fn.get().is_some() {
        Some(&live_leader)
    } else {
        None
    };
    resolve_decision(
        vshard_id,
        state.node_id,
        routing_guard.as_deref(),
        live_lookup,
    )
}

/// Parameters for [`dispatch_superstep_to_node`].
pub(in crate::control::server::graph_dispatch) struct DispatchSuperstepParams<'a> {
    pub(in crate::control::server::graph_dispatch) tenant_id: TenantId,
    pub(in crate::control::server::graph_dispatch) database_id: DatabaseId,
    pub(in crate::control::server::graph_dispatch) deadline_ms: u64,
    pub(in crate::control::server::graph_dispatch) node_id: u64,
    pub(in crate::control::server::graph_dispatch) is_local: bool,
    pub(in crate::control::server::graph_dispatch) route_vshard: u32,
    pub(in crate::control::server::graph_dispatch) plan: PhysicalPlan,
    pub(in crate::control::server::graph_dispatch) version_set: &'a GatewayVersionSet,
}

/// Dispatch a single already-built graph-superstep `plan` to one owner node and
/// return its node-level payload. The LOCAL node fans the plan across all its
/// Data-Plane cores via `execute_plan_all_local_cores` (per-core results merged
/// into one payload); a REMOTE node gets one `RouteDecision::Remote` dispatch via
/// `dispatch_route`. An empty payload denotes a zero-vertex shard — the caller's
/// decoder maps it to its result type's `::default()`. Shared by the PageRank and
/// WCC per-node scatter paths.
pub(in crate::control::server::graph_dispatch) async fn dispatch_superstep_to_node(
    shared_arc: &Arc<SharedState>,
    args: DispatchSuperstepParams<'_>,
) -> crate::Result<Payload> {
    let DispatchSuperstepParams {
        tenant_id,
        database_id,
        deadline_ms,
        node_id,
        is_local,
        route_vshard,
        plan,
        version_set,
    } = args;
    if is_local {
        // Local node: fan across ALL local cores and merge. The per-core
        // owned-node sets are disjoint, so the merged result is correct without
        // dedup. At 1 core/node this is behaviour-identical to a single-core
        // dispatch.
        let node_result = execute_plan_all_local_cores(
            shared_arc.as_ref(),
            tenant_id,
            database_id,
            plan,
            TraceId::ZERO,
            // This resolve path carries no session-transaction context.
            None,
        )
        .await?;
        Ok(Payload::from_vec(node_result.payload))
    } else {
        // Remote node: one dispatch via the gateway.
        let route = TaskRoute {
            plan,
            decision: RouteDecision::Remote {
                node_id,
                vshard_id: route_vshard as u64,
            },
            vshard_id: route_vshard,
        };
        let payloads = dispatch_route(DispatchRouteParams {
            route,
            shared: shared_arc,
            tenant_id,
            database_id,
            trace_id: TraceId::ZERO,
            deadline_ms,
            version_set,
            // This resolve path carries no session-transaction context.
            txn_id: None,
        })
        .await?
        .payloads;
        payloads
            .into_iter()
            .next()
            .map(Payload::from_vec)
            .ok_or_else(|| crate::Error::Internal {
                detail: format!("graph superstep: node={node_id} returned no payload"),
            })
    }
}

/// The gateway's `Arc<SharedState>` for the remote dispatch path. In cluster
/// mode the gateway is always wired; failing loudly here beats silently
/// degrading to a local-only (partial) scatter.
///
/// `pub(crate)` so the in-transaction staging choke points
/// (`session::leader_forward`) can obtain the `Arc<SharedState>` the remote
/// dispatch primitive requires when forwarding a staged write / overlay drop to
/// a remote leader.
pub(crate) fn gateway_shared(state: &SharedState) -> crate::Result<Arc<SharedState>> {
    let gateway = state.gateway.get().ok_or_else(|| crate::Error::Internal {
        detail: "graph scatter: cluster routing present but gateway unavailable for \
                 remote dispatch"
            .into(),
    })?;
    // Upgrade the gateway's `Weak<SharedState>` back-reference. Always
    // succeeds while the node runs; a `None` (racing full teardown) surfaces
    // as the accessor's own typed shutdown error.
    gateway.shared()
}