netget/server/bfd/
runner.rs

1//! Runs one [`Session`] against the network: periodic transmission with jitter, the Final
2//! answer to a Poll, the Detection Time, and commands from the model or the dashboard.
3//! The server and the client each run one of these per session; the model is consulted
4//! outside it, through the [`Note`]s it sends.
5use super::packet::{self, ControlPacket, State};
6use super::session::{Session, Timers};
7use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
8use anyhow::{bail, Context, Result};
9use serde_json::{json, Value};
10use std::net::SocketAddr;
11use std::sync::Arc;
12use std::time::Duration;
13use tokio::net::UdpSocket;
14use tokio::sync::mpsc;
15use tokio::time::Instant;
16
17pub enum Input {
18    Packet {
19        bytes: Vec<u8>,
20        packet: ControlPacket,
21    },
22    /// An action for this session. `depth` is how many model turns led to it; `command` is
23    /// the injected caller waiting for an outcome, if any.
24    Action {
25        action: Value,
26        depth: u32,
27        command: Option<ClientCommand>,
28    },
29}
30
31pub enum Note {
32    Changed {
33        previous: State,
34        snapshot: Value,
35        depth: u32,
36    },
37    /// The session ended (idle, or its inputs closed).
38    Ended { local_discr: u32 },
39}
40
41/// Validate a session action: `bfd_set_timers`, `bfd_admin_down`, `bfd_admin_up`.
42pub fn check_action(v: &Value) -> Result<()> {
43    match v["type"].as_str().unwrap_or_default() {
44        "bfd_set_timers" => {
45            Timers::from_json(v, Timers::default())?;
46        }
47        "bfd_admin_down" => {
48            diag_of(v)?;
49        }
50        "bfd_admin_up" => {}
51        other => bail!("{other:?} is not a BFD session action"),
52    }
53    Ok(())
54}
55
56fn diag_of(v: &Value) -> Result<u8> {
57    match v["diag"].as_str() {
58        None => Ok(7),
59        Some(name) => packet::diag_code(name),
60    }
61}
62
63fn apply(s: &mut Session, v: &Value) -> Result<Option<(State, State)>> {
64    check_action(v)?;
65    Ok(match v["type"].as_str().unwrap_or_default() {
66        "bfd_set_timers" => {
67            s.set_timers(Timers::from_json(v, s.configured)?);
68            None
69        }
70        "bfd_admin_down" => s.admin_down(diag_of(v)?),
71        _ => s.admin_up(),
72    })
73}
74
75pub struct Link {
76    pub socket: Arc<UdpSocket>,
77    pub dest: SocketAddr,
78    pub label: String,
79}
80
81impl Link {
82    async fn send(&self, s: &mut Session, final_: bool) {
83        let bytes = s.packet(final_);
84        if let Err(e) = self.socket.send_to(&bytes, self.dest).await {
85            tracing::warn!("{}: sending to {} failed: {e}", self.label, self.dest);
86        }
87    }
88}
89
90/// Run until the inputs close, or — when `idle` is given — until the session has been Down
91/// with nothing heard from the peer for that long.
92pub async fn run(
93    mut s: Session,
94    link: Link,
95    mut inputs: mpsc::Receiver<Input>,
96    notes: mpsc::Sender<Note>,
97    idle: Option<Duration>,
98) {
99    let started = Instant::now();
100    let mut next_tx = Instant::now();
101    loop {
102        let detect = s.detection_deadline();
103        let idle_at = idle.and_then(|d| {
104            matches!(s.state, State::Down | State::AdminDown)
105                .then(|| s.last_rx.unwrap_or(started) + d)
106        });
107        let (transition, depth) = tokio::select! {
108            input = inputs.recv() => match input {
109                None => break,
110                Some(Input::Packet { bytes, packet }) => {
111                    let could_send = s.may_transmit();
112                    match s.receive(&bytes, &packet, Instant::now()) {
113                        Ok(r) => {
114                            if r.send_final {
115                                link.send(&mut s, true).await;
116                            }
117                            if r.transition.is_some() || !could_send {
118                                next_tx = Instant::now();
119                            }
120                            (r.transition, 0)
121                        }
122                        Err(e) => {
123                            tracing::warn!("{} dropped a packet decision=dropped: {e:#}", link.label);
124                            continue;
125                        }
126                    }
127                }
128                Some(Input::Action { action, depth, command }) => match apply(&mut s, &action) {
129                    Ok(t) => {
130                        // On the wire before the caller hears it was done.
131                        if s.may_transmit() {
132                            link.send(&mut s, false).await;
133                        }
134                        next_tx = Instant::now() + s.next_interval();
135                        if let Some(c) = command {
136                            crate::client::command_support::reply(c, Ok(ClientSendOutcome::Executed {
137                                detail: s.snapshot().to_string(),
138                            }));
139                        }
140                        (t, depth)
141                    }
142                    Err(e) => {
143                        tracing::warn!("{} refused {}: {e:#}", link.label, action["type"]);
144                        if let Some(c) = command {
145                            crate::client::command_support::reply(c, Ok(ClientSendOutcome::Rejected {
146                                error: e.to_string(),
147                            }));
148                        }
149                        continue;
150                    }
151                },
152            },
153            _ = tokio::time::sleep_until(next_tx) => {
154                if s.may_transmit() {
155                    link.send(&mut s, false).await;
156                }
157                next_tx = Instant::now() + s.next_interval();
158                continue;
159            }
160            _ = sleep_until_opt(detect) => (s.detection_expired(), 0),
161            _ = sleep_until_opt(idle_at) => {
162                tracing::info!("{}: Down with nothing heard for {:?}; ending the session", link.label, idle.unwrap_or_default());
163                break;
164            }
165        };
166        if let Some((previous, _)) = transition {
167            let note = Note::Changed {
168                previous,
169                snapshot: s.snapshot(),
170                depth,
171            };
172            if !notes.is_closed() && notes.try_send(note).is_err() {
173                tracing::warn!(
174                    "{}: a state change was not reported: the model is behind decision=note_queue_full",
175                    link.label
176                );
177            }
178        }
179    }
180    let _ = notes.try_send(Note::Ended {
181        local_discr: s.local_discr,
182    });
183}
184
185async fn sleep_until_opt(at: Option<Instant>) {
186    match at {
187        Some(t) => tokio::time::sleep_until(t).await,
188        None => std::future::pending().await,
189    }
190}
191
192/// A tx socket bound to `ip` on a source port in 49152–65535 (RFC 5881 §4), TTL 255.
193pub async fn tx_socket(ip: std::net::IpAddr) -> Result<UdpSocket> {
194    let mut last = None;
195    for _ in 0..64 {
196        let port = packet::SOURCE_PORTS.start()
197            + (rand::random::<u16>() % (packet::SOURCE_PORTS.end() - packet::SOURCE_PORTS.start()));
198        match UdpSocket::bind(SocketAddr::new(ip, port)).await {
199            Ok(s) => {
200                if ip.is_ipv4() {
201                    s.set_ttl(255).context("setting TTL 255")?;
202                }
203                return Ok(s);
204            }
205            Err(e) => last = Some(e),
206        }
207    }
208    Err(last
209        .map(anyhow::Error::from)
210        .unwrap_or_else(|| anyhow::anyhow!("no source port")))
211    .context("no free BFD source port in 49152-65535")
212}
213
214/// The event data for a state change, given who the peer is.
215pub fn state_event(peer: &str, previous: State, snapshot: &Value) -> Value {
216    let mut data = snapshot.clone();
217    data["peer"] = json!(peer);
218    data["previous_state"] = json!(previous.name());
219    data
220}
netget/server/bfd/runner.rs

netget/server/bfd/
runner.rs
