vox 0.4.0

Core Vox library crate
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
158
159

//! Tests for ConnectBuilder::wait_for_service initial-connect waiting.

use std::time::Duration;
use vox_core::SessionError;

#[vox::service]
trait Echo {
    async fn echo(&self, value: u32) -> u32;
}

#[derive(Clone)]
struct EchoService;

impl Echo for EchoService {
    async fn echo(&self, value: u32) -> u32 {
        value
    }
}

// r[verify session.initial-connect-waiting]
// r[verify session.initial-connect-waiting.retryable]
// r[verify session.initial-connect-waiting.timeout]
#[cfg(unix)]
#[tokio::test]
async fn wait_for_service_retries_until_service_appears() {
    let dir = tempfile::tempdir().expect("tempdir");
    let sock_path = dir.path().join("echo.sock");
    let addr = format!("local://{}", sock_path.display());

    // Socket does not exist yet. Server starts after a short delay.
    let server = {
        let addr = addr.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(50)).await;
            vox::serve(&addr, EchoDispatcher::new(EchoService))
                .await
                .expect("serve");
        })
    };

    let client: EchoClient = vox::connect(&addr)
        .connect_timeout(Duration::from_millis(100))
        .wait_for_service(Duration::from_secs(5))
        .await
        .expect("should connect once service starts");

    let result = client.echo(42).await.expect("echo");
    assert_eq!(result, 42);

    server.abort();
}

// r[verify session.initial-connect-waiting]
// r[verify session.initial-connect-waiting.retryable]
// r[verify session.initial-connect-waiting.timeout]
// r[verify session.initial-connect-waiting.backoff]
// r[verify session.initial-connect-waiting.no-session]
#[tokio::test]
async fn wait_for_service_times_out_when_service_never_starts() {
    // Bind to get a free port then drop the listener so nothing is listening.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("bind");
    let addr = listener.local_addr().expect("local_addr");
    drop(listener);

    let result = vox::connect::<EchoClient>(format!("tcp://{addr}"))
        .connect_timeout(Duration::from_millis(50))
        .wait_for_service(Duration::from_millis(200))
        .establish()
        .await;

    let err = match result {
        Err(e) => e,
        Ok(_) => panic!("expected failure when service never starts"),
    };
    assert!(
        matches!(err, SessionError::Io(_) | SessionError::ConnectTimeout),
        "error should be retryable kind (Io or ConnectTimeout), got: {err:?}"
    );
}

// r[verify session.initial-connect-waiting.timeout]
#[tokio::test]
async fn wait_for_service_deadline_caps_individual_attempt() {
    // Verifies that a single slow attempt cannot exceed the waiting timeout.
    // connect_timeout (5s) is much larger than wait_for_service (200ms).
    // Without the fix, a single attempt would block for up to 5s, violating
    // the 200ms budget. With the fix, the attempt is capped by remaining budget.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("bind");
    let addr = listener.local_addr().expect("local_addr");

    // Accept connections but never respond — causes connect_timeout to fire.
    let server = tokio::spawn(async move {
        loop {
            let _ = listener.accept().await;
            // hold the socket open without doing anything
            tokio::time::sleep(Duration::from_secs(60)).await;
        }
    });

    let start = std::time::Instant::now();
    let result = vox::connect::<EchoClient>(format!("tcp://{addr}"))
        .connect_timeout(Duration::from_secs(5)) // much larger than wait_for_service
        .wait_for_service(Duration::from_millis(200))
        .establish()
        .await;
    let elapsed = start.elapsed();
    server.abort();

    assert!(result.is_err(), "should fail");
    assert!(
        elapsed < Duration::from_secs(2),
        "waiting timeout must bound individual attempts; elapsed: {elapsed:?}"
    );
}

// r[verify session.initial-connect-waiting.non-retryable]
#[tokio::test]
async fn wait_for_service_fails_immediately_on_protocol_error() {
    // Server that immediately closes the connection without speaking vox.
    // This causes SessionError::Protocol (link closed during transport prologue),
    // which is non-retryable and must not consume the full wait_for_service timeout.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("bind");
    let addr = listener.local_addr().expect("local_addr");

    let server = tokio::spawn(async move {
        loop {
            let (socket, _) = listener.accept().await.expect("accept");
            drop(socket);
        }
    });

    let start = std::time::Instant::now();
    let result = vox::connect::<EchoClient>(format!("tcp://{addr}"))
        .connect_timeout(Duration::from_millis(100))
        .wait_for_service(Duration::from_secs(10)) // long timeout — should NOT be consumed
        .establish()
        .await;
    let elapsed = start.elapsed();
    server.abort();

    let err = match result {
        Err(e) => e,
        Ok(_) => panic!("should fail with protocol error"),
    };
    assert!(
        elapsed < Duration::from_secs(2),
        "non-retryable error should fail fast, not wait the full timeout; elapsed: {elapsed:?}"
    );
    assert!(
        matches!(err, SessionError::Protocol(_)),
        "expected SessionError::Protocol for peer that closed connection, got: {err:?}"
    );
}