aurelia 0.1.0

Embeddable service mesh for Rust distributed applications.
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

// This file is part of the Aurelia workspace.
// SPDX-FileCopyrightText: 2026 Zivatar Limited
// SPDX-License-Identifier: Apache-2.0

use super::*;
use crate::peering::observability::ObservabilityHandle;

#[allow(clippy::too_many_arguments)]
pub(super) async fn ensure_primary_dial<B>(
    session: &Arc<PeerSession>,
    state: &mut PeerState,
    dial_addr: &Arc<Mutex<Option<DomusAddr>>>,
    backend: &Arc<B>,
    config: &DomusConfigAccess,
    blob: &Arc<BlobManager>,
    registry: &Arc<TabernaRegistry>,
    peer_state_tx: &mpsc::Sender<PeerStateUpdate>,
    primary_available: &Arc<Notify>,
    primary_dispatch: &Arc<PrimaryDispatchQueue>,
    callis_tracker: &CallisTracker,
    observability: &ObservabilityHandle,
    has_pending: bool,
) where
    B: TransportBackend<Addr = DomusAddr> + 'static,
{
    if !state.primary.is_empty() || state.dialing_primary {
        return;
    }
    if !should_reconnect_primary(session, state, has_pending).await {
        return;
    }
    let max_parallel = config.snapshot().await.max_parallel_callis_per_peer.max(1);
    if callis_tracker.count() >= max_parallel {
        return;
    }
    state.dialing_primary = true;
    let delay = if state.role == PeerRole::Listener {
        compute_listener_delay(config, state.role, state.had_primary).await
    } else {
        compute_reconnect_delay(config, state.reconnect_attempt).await
    };
    state.reconnect_attempt = state.reconnect_attempt.saturating_add(1);
    if let Some(addr) = current_dial_addr(dial_addr).await {
        let runtime_handle = session.runtime_handle();
        spawn_dial_task(
            addr,
            Arc::clone(backend),
            config.clone(),
            session.clone(),
            blob.clone(),
            registry.clone(),
            delay,
            peer_state_tx.clone(),
            Arc::clone(primary_available),
            Arc::clone(primary_dispatch),
            callis_tracker.clone(),
            observability.clone(),
            runtime_handle,
        );
    }
    if state.role == PeerRole::Listener {
        state.role = PeerRole::Originator;
    }
}

pub(super) fn remove_primary_handle(
    handles: &mut VecDeque<CallisHandle>,
    id: CallisId,
) -> Option<CallisHandle> {
    let pos = handles.iter().position(|handle| handle.id == id)?;
    handles.remove(pos)
}

pub(super) async fn compute_listener_delay(
    config: &DomusConfigAccess,
    role: PeerRole,
    had_primary: bool,
) -> Duration {
    let cfg = config.snapshot().await;
    match role {
        PeerRole::Originator => Duration::from_secs(0),
        PeerRole::Listener => {
            if had_primary {
                cfg.listener_reconnect_timeout
            } else {
                cfg.listener_delay
            }
        }
    }
}

pub(super) async fn compute_reconnect_delay(
    config: &DomusConfigAccess,
    attempt: usize,
) -> Duration {
    if attempt == 0 {
        return Duration::from_secs(0);
    }
    let cfg = config.snapshot().await;
    if cfg.reconnect_backoff.is_empty() {
        return Duration::from_secs(4);
    }
    let idx = attempt.saturating_sub(1);
    if idx >= cfg.reconnect_backoff.len() {
        *cfg.reconnect_backoff
            .last()
            .unwrap_or(&Duration::from_secs(4))
    } else {
        cfg.reconnect_backoff[idx]
    }
}

pub(super) async fn should_reconnect_primary(
    session: &Arc<PeerSession>,
    state: &PeerState,
    has_pending: bool,
) -> bool {
    if state.closing || session.is_closing() {
        return false;
    }
    if has_pending {
        return true;
    }
    session.has_inflight().await
}

pub(super) async fn current_dial_addr(
    dial_addr: &Arc<Mutex<Option<DomusAddr>>>,
) -> Option<DomusAddr> {
    let guard = dial_addr.lock().await;
    guard.clone()
}