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

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

//! Velo-specific integration tests that cannot live in the shared macro suite.
//! TEST-03: Exclusive reject — requires multi_thread runtime and concurrent attach.
//! TEST-07: Transport error propagation — tests closed transport receiver path.

mod common;

use common::MockFrameTransport;
use futures::StreamExt;
use std::sync::Arc;
use std::time::Duration;
use velo::streaming::{AnchorManager, AttachError, StreamError, StreamFrame};
use velo_ext::WorkerId;

fn make_manager() -> Arc<AnchorManager> {
    Arc::new(AnchorManager::new(
        WorkerId::from_u64(1),
        Arc::new(MockFrameTransport::new()),
    ))
}

// TEST-03: Exclusive reject
// Requires multi_thread flavor to expose true concurrency in DashMap::entry() path.
#[tokio::test(flavor = "multi_thread")]
async fn test_03_exclusive_reject() {
    let mgr = make_manager();
    let anchor = mgr.create_anchor::<u32>();
    let handle = anchor.handle();

    // First attach succeeds
    let _sender = mgr
        .attach_stream_anchor::<u32>(handle)
        .await
        .expect("first attach must succeed");

    // Second attach while first is active must return AlreadyAttached
    let result = mgr.attach_stream_anchor::<u32>(handle).await;
    assert!(
        matches!(result, Err(AttachError::AlreadyAttached { .. })),
        "concurrent attach must return AlreadyAttached, got {:?}",
        result
    );
}

// TEST-07: Transport error propagation
// Strategy: Attach a sender so the stream is active, then drop the sender without
// explicit detach/finalize. impl Drop sends StreamFrame::Dropped synchronously,
// which StreamAnchor maps to Err(StreamError::SenderDropped) — stream terminates.
//
// This validates the transport-close path: when the sender side disappears
// (network drop simulated by sender drop), the consumer-side StreamAnchor
// observes the termination sentinel and stops yielding items.
#[tokio::test(flavor = "multi_thread")]
async fn test_07_transport_error_propagation() {
    let mgr = make_manager();

    let mut anchor = mgr.create_anchor::<u32>();
    let handle = anchor.handle();
    // Attach a sender to start the stream.
    let sender = mgr
        .attach_stream_anchor::<u32>(handle)
        .await
        .expect("attach must succeed");

    // Send one item to confirm normal delivery.
    sender.send(42u32).await.expect("send must succeed");

    // Drop the sender WITHOUT finalize/detach — simulates transport closure.
    // impl Drop sends StreamFrame::Dropped synchronously before returning.
    drop(sender);

    // Stream must deliver the item, then the Dropped sentinel, then None.
    let frame1 = tokio::time::timeout(Duration::from_secs(5), anchor.next())
        .await
        .expect("stream must resolve within 5s");
    assert!(
        matches!(frame1, Some(Ok(StreamFrame::Item(42u32)))),
        "first frame must be Item(42), got {:?}",
        frame1
    );

    // Stream yields Err(SenderDropped) for the Dropped sentinel.
    let frame2 = tokio::time::timeout(Duration::from_secs(5), anchor.next())
        .await
        .expect("stream must resolve within 5s");
    assert!(
        matches!(frame2, Some(Err(StreamError::SenderDropped))),
        "second frame must be Err(SenderDropped), got {:?}",
        frame2
    );

    // Stream must yield None after the terminal sentinel.
    assert!(
        anchor.next().await.is_none(),
        "stream must be exhausted after SenderDropped"
    );
}