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

// SPDX-License-Identifier: BUSL-1.1

//! Per-node `WccSuperstep` dispatch: one plan per distinct owner node, all
//! issued concurrently via `join_all`, decoding each node's
//! [`WccSuperstepResult`].
//!
//! Each dispatch carries the owner node's FULL `owned_vshards` set so the
//! handler contracts every node homed on that owner in a single CSR pass. A
//! remote node gets one `RouteDecision::Remote` dispatch; the local node fans
//! across ALL local cores via `execute_plan_all_local_cores` (which merges the
//! per-core disjoint results into one `WccSuperstepResult` before returning).
//! At 1 core/node the fan is over a single core and behaviour is identical to a
//! single-core dispatch.

use futures::future::join_all;

use crate::bridge::envelope::{Payload, PhysicalPlan};
use crate::control::gateway::version_set::GatewayVersionSet;
use crate::control::server::graph_dispatch::bsp_pagerank::enumerate::ShardTarget;
use crate::control::server::graph_dispatch::cluster_resolve::{
    DispatchSuperstepParams, dispatch_superstep_to_node, gateway_shared,
};
use crate::types::{DatabaseId, TenantId};
use nodedb_graph::AlgoParams;
use nodedb_physical::physical_plan::{GraphOp, WccSuperstepPlan, WccSuperstepResult};

/// One owner node's decoded WCC result.
pub(super) struct ShardWccResult {
    pub(super) result: WccSuperstepResult,
}

/// Dispatch one `WccSuperstep` to every owner node concurrently and decode each
/// node's [`WccSuperstepResult`]. Single round — there is no count phase or
/// loop; the coordinator stitches the returned results globally.
pub(super) async fn scatter_wcc_round(
    state: &crate::control::state::SharedState,
    tenant_id: TenantId,
    database_id: DatabaseId,
    params: &AlgoParams,
    targets: &[ShardTarget],
    deadline_ms: u64,
) -> crate::Result<Vec<ShardWccResult>> {
    let shared_arc = gateway_shared(state)?;
    let version_set = GatewayVersionSet::from_pairs(Vec::new());

    let futs = targets.iter().map(|t| {
        let plan = PhysicalPlan::Graph(GraphOp::WccSuperstep(Box::new(WccSuperstepPlan {
            params: params.clone(),
            // FULL owned set for this node — the handler contracts every node
            // homed here in one pass and emits boundary edges only for dsts on
            // OTHER nodes.
            owned_vshards: t.owned_vshards.clone(),
        })));
        let version_set = version_set.clone();
        let node_id = t.node_id;
        let is_local = t.is_local;
        let route_vshard = t.route_vshard();
        let shared_arc = shared_arc.clone();

        Box::pin(async move {
            let payload = dispatch_superstep_to_node(
                &shared_arc,
                DispatchSuperstepParams {
                    tenant_id,
                    database_id,
                    deadline_ms,
                    node_id,
                    is_local,
                    route_vshard,
                    plan,
                    version_set: &version_set,
                },
            )
            .await?;
            let result = decode_wcc_from_payload(node_id, payload)?;
            Ok::<ShardWccResult, crate::Error>(ShardWccResult { result })
        })
    });

    let results = join_all(futs).await;
    let mut out = Vec::with_capacity(results.len());
    for res in results {
        out.push(res?);
    }
    Ok(out)
}

/// Decode a single node's `WccSuperstepResult` from an already-wrapped
/// [`Payload`]. Empty payload → `WccSuperstepResult::default()` (zero-vertex
/// shard, contributes no labels or boundary edges).
fn decode_wcc_from_payload(node_id: u64, payload: Payload) -> crate::Result<WccSuperstepResult> {
    if payload.is_empty() {
        return Ok(WccSuperstepResult::default());
    }
    zerompk::from_msgpack::<WccSuperstepResult>(payload.as_ref()).map_err(|e| crate::Error::Codec {
        detail: format!("wcc: node={node_id} result decode: {e}"),
    })
}