velo 0.4.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

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

//! System handler registration for distributed event messages.

use std::sync::Arc;

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

use super::VeloEvents;

/// Register the three event system handlers with the handler manager.
///
/// These are fire-and-forget active messages — the handler return value has no
/// observable effect on the caller, so errors are intentionally logged and
/// discarded rather than propagated. For `_event_trigger_request`, ACK/NACK
/// semantics are managed internally via response channels in `handle_trigger_request`,
/// not through the handler's `Result`.
pub(crate) fn register_event_handlers(
    handlers: &HandlerManager,
    events: Arc<VeloEvents>,
) -> anyhow::Result<()> {
    // _event_subscribe: Remote node subscribes to a local event
    let events_clone = events.clone();
    handlers.register_internal_handler(
        Handler::am_handler_async("_event_subscribe", move |ctx| {
            let events = events_clone.clone();
            async move {
                if let Err(e) = events.handle_subscribe(ctx.payload).await {
                    tracing::warn!("_event_subscribe handler error: {}", e);
                }
                Ok(())
            }
        })
        .spawn()
        .build(),
    )?;

    // _event_trigger: Completion notification from remote owner to subscriber
    let events_clone = events.clone();
    handlers.register_internal_handler(
        Handler::am_handler("_event_trigger", move |ctx| {
            if let Err(e) = events_clone.handle_trigger(ctx.payload) {
                tracing::warn!("_event_trigger handler error: {}", e);
            }
            Ok(())
        })
        .spawn()
        .build(),
    )?;

    // _event_trigger_request: Remote trigger/poison request + ACK/NACK
    let events_clone = events.clone();
    handlers.register_internal_handler(
        Handler::am_handler_async("_event_trigger_request", move |ctx| {
            let events = events_clone.clone();
            async move {
                if let Err(e) = events.handle_trigger_request(ctx.payload).await {
                    tracing::warn!("_event_trigger_request handler error: {}", e);
                }
                Ok(())
            }
        })
        .spawn()
        .build(),
    )?;

    Ok(())
}