velo 0.12.0

Velo distributed-systems runtime: active messaging, peer discovery, streaming, rendezvous, and queue backends
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

// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! System handlers for messenger active message infrastructure.

use anyhow::Result;
use serde::{Deserialize, Serialize};
use velo_ext::PeerInfo;

use crate::messenger::handlers::{Handler, HandlerManager, TypedContext};

/// Request payload for the _hello handshake handler
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HelloRequest {
    /// The peer information of the sender
    pub peer_info: PeerInfo,
}

/// Response payload for handler list queries
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HandlersResponse {
    /// List of available handler names
    pub handlers: Vec<String>,
}

/// Create the _hello system handler.
fn create_hello_handler() -> Handler {
    Handler::typed_unary_async("_hello", |ctx: TypedContext<HelloRequest>| async move {
        let peer_info = ctx.input.peer_info;

        tracing::debug!(
            target: "crate::messenger::system",
            instance_id = %peer_info.instance_id(),
            "Received _hello handshake from peer"
        );

        ctx.msg.register_peer(peer_info.clone())?;

        let handlers = ctx.msg.list_local_handlers();

        tracing::debug!(
            target: "crate::messenger::system",
            instance_id = %peer_info.instance_id(),
            handler_count = handlers.len(),
            "Completed _hello handshake"
        );

        Ok(HandlersResponse { handlers })
    })
    .spawn()
    .build()
}

/// Create the _list_handlers system handler.
fn create_list_handlers_handler() -> Handler {
    Handler::typed_unary_async("_list_handlers", |ctx: TypedContext<()>| async move {
        let handlers = ctx.msg.list_local_handlers();

        tracing::debug!(
            target: "crate::messenger::system",
            handler_count = handlers.len(),
            "Responding to _list_handlers query"
        );

        Ok(HandlersResponse { handlers })
    })
    .spawn()
    .build()
}

/// Register all system handlers with the dispatcher.
pub(crate) fn register_system_handlers(manager: &HandlerManager) -> Result<()> {
    manager.register_internal_handler(create_hello_handler())?;
    manager.register_internal_handler(create_list_handlers_handler())?;

    tracing::info!(
        target: "crate::messenger::system",
        "Registered system handlers: _hello, _list_handlers"
    );

    Ok(())
}

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

    /// Create an empty test address for testing
    fn make_empty_test_address() -> velo_ext::WorkerAddress {
        let map: HashMap<String, Vec<u8>> = HashMap::new();
        let encoded = rmp_serde::to_vec(&map).unwrap();
        velo_ext::WorkerAddress::from_encoded(encoded)
    }

    #[test]
    fn test_hello_request_serialization() {
        use velo_ext::InstanceId;

        let instance_id = InstanceId::new_v4();
        let address = make_empty_test_address();
        let peer_info = PeerInfo::new(instance_id, address);

        let request = HelloRequest {
            peer_info: peer_info.clone(),
        };

        let json = serde_json::to_string(&request).unwrap();
        let deserialized: HelloRequest = serde_json::from_str(&json).unwrap();

        assert_eq!(
            request.peer_info.instance_id(),
            deserialized.peer_info.instance_id()
        );
    }

    #[test]
    fn test_handlers_response_serialization() {
        let response = HandlersResponse {
            handlers: vec![
                "handler1".to_string(),
                "handler2".to_string(),
                "_system".to_string(),
            ],
        };

        let json = serde_json::to_string(&response).unwrap();
        let deserialized: HandlersResponse = serde_json::from_str(&json).unwrap();

        assert_eq!(response.handlers.len(), deserialized.handlers.len());
        assert_eq!(response.handlers, deserialized.handlers);
    }
}