netget/server/bfd/
mod.rs

1//! BFD speaker (RFC 5880 Asynchronous mode over UDP: RFC 5881 single-hop, RFC 5883
2//! multihop), in the passive role: a peer that starts sending is a session request for the
3//! model; an accepted session is run in Rust (`runner.rs`) and every state change is the
4//! model's to react to. **Deliberately silent**: BFD has no refusal, so a peer that is not
5//! accepted — or a backend failure — is never answered and its session stays Down.
6pub mod actions;
7pub mod packet;
8pub mod runner;
9pub mod session;
10pub mod ttl;
11
12use crate::llm::action_helper::call_llm;
13use crate::llm::actions::protocol_trait::ActionResult;
14use crate::logging::emit::Log;
15use crate::protocol::{Event, SpawnContext};
16use crate::server::connection::ConnectionId;
17use actions::BfdProtocol;
18use anyhow::{Context, Result};
19use packet::{AuthConfig, ControlPacket};
20use runner::{Input, Link, Note};
21use serde_json::{json, Value};
22use session::{Session, Timers};
23use std::collections::{HashMap, HashSet};
24use std::net::{IpAddr, SocketAddr};
25use std::sync::{Arc, Mutex};
26use std::time::Duration;
27use tokio::net::UdpSocket;
28use tokio::sync::mpsc;
29use tokio::time::Instant;
30
31/// A session that is Down with nothing heard from its peer for this long ends.
32pub const SESSION_IDLE: Duration = Duration::from_secs(60);
33/// A peer the model did not accept is not asked about again for this long (a Down peer
34/// sends once a second, and each packet would otherwise be a model call).
35pub const DECLINE_HOLD: Duration = Duration::from_secs(30);
36/// How many model turns may follow from one another on one session.
37pub const MAX_FOLLOWUP_DEPTH: u32 = 8;
38const SESSION_INPUTS: usize = 64;
39const NOTES: usize = 64;
40
41struct Entry {
42    peer: IpAddr,
43    inputs: mpsc::Sender<Input>,
44    conn: ConnectionId,
45}
46
47#[derive(Default)]
48struct Table {
49    by_discr: HashMap<u32, Entry>,
50    by_peer: HashMap<IpAddr, u32>,
51    asking: HashSet<IpAddr>,
52    declined: HashMap<IpAddr, Instant>,
53}
54
55impl Table {
56    fn new_discriminator(&self) -> u32 {
57        loop {
58            let d: u32 = rand::random();
59            if d != 0 && !self.by_discr.contains_key(&d) {
60                return d;
61            }
62        }
63    }
64}
65
66#[derive(Clone)]
67struct Shared {
68    ctx: SpawnContext,
69    table: Arc<Mutex<Table>>,
70    tx: Arc<UdpSocket>,
71    port: u16,
72    multihop: bool,
73    max_sessions: usize,
74    timers: Timers,
75    auth: Option<AuthConfig>,
76}
77
78fn lock(t: &Mutex<Table>) -> std::sync::MutexGuard<'_, Table> {
79    t.lock().unwrap_or_else(|e| e.into_inner())
80}
81
82pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
83    let params = ctx.startup_params.as_ref();
84    let port = ctx.port.unwrap_or_else(|| ctx.legacy_listen_addr().port());
85    let host = ctx.host.clone().unwrap_or_else(|| "127.0.0.1".into());
86    let bind = tokio::net::lookup_host((host.as_str(), port))
87        .await
88        .with_context(|| format!("cannot resolve host {host}"))?
89        .next()
90        .with_context(|| format!("host {host} resolved to no address"))?;
91    let multihop = params
92        .map(|p| p.get_optional_bool("multihop"))
93        .transpose()?
94        .flatten()
95        .unwrap_or(port == packet::MULTIHOP_PORT);
96    let max_sessions = params
97        .map(|p| p.get_optional_u64("max_sessions"))
98        .transpose()?
99        .flatten()
100        .map(|n| n as usize)
101        .unwrap_or(actions::MAX_SESSIONS);
102    let timers = actions::timers_from(params)?;
103    let auth = actions::auth_from(params)?;
104    // SO_REUSEADDR: a routing daemon on the same host binds the BFD port on the wildcard
105    // address, and a socket on a specific address still receives what is sent to it.
106    let socket = crate::server::socket_helpers::create_reusable_udp_socket(bind)
107        .await
108        .with_context(|| format!("BFD failed to bind {bind}"))?;
109    ttl::enable(&socket).context("asking for the TTL of received packets")?;
110    let local = socket.local_addr()?;
111    let tx = Arc::new(runner::tx_socket(local.ip()).await?);
112    let log = Log::new(Some(&ctx.status_tx));
113    log.info(format!(
114        "BFD listening on {local} ({}), sending from {}",
115        if multihop {
116            "multihop, RFC 5883"
117        } else {
118            "single-hop, RFC 5881: TTL 255 required"
119        },
120        tx.local_addr()?
121    ));
122    let shared = Shared {
123        ctx: ctx.clone(),
124        table: Default::default(),
125        tx,
126        port: local.port(),
127        multihop,
128        max_sessions,
129        timers,
130        auth,
131    };
132    let task = tokio::spawn(async move {
133        let mut buf = [0u8; 2048];
134        loop {
135            let (n, from, ttl) = match ttl::recv(&socket, &mut buf).await {
136                Ok(r) => r,
137                Err(e) => {
138                    Log::new(Some(&shared.ctx.status_tx)).error(format!("BFD receive error: {e}"));
139                    break;
140                }
141            };
142            if let Some(asking) = receive(&shared, &buf[..n], from, ttl).await {
143                shared
144                    .ctx
145                    .state
146                    .register_server_task(shared.ctx.server_id, asking)
147                    .await;
148            }
149        }
150    });
151    ctx.state.register_server_task(ctx.server_id, task).await;
152    Ok(local)
153}
154
155/// Route one datagram. A new peer starts a model turn, whose task is returned to register.
156async fn receive(
157    shared: &Shared,
158    bytes: &[u8],
159    from: SocketAddr,
160    ttl: Option<u8>,
161) -> Option<tokio::task::JoinHandle<()>> {
162    if !shared.multihop && ttl.is_some_and(|t| t != 255) {
163        tracing::warn!(
164            "BFD dropped a packet from {from} with TTL {} (single-hop requires 255) decision=dropped_ttl",
165            ttl.unwrap_or_default()
166        );
167        return None;
168    }
169    let packet = match packet::decode(bytes) {
170        Ok(p) => p,
171        Err(e) => {
172            tracing::warn!("BFD dropped a datagram from {from}: {e:#}");
173            return None;
174        }
175    };
176    let peer = from.ip();
177    let input = || Input::Packet {
178        bytes: bytes.to_vec(),
179        packet: packet.clone(),
180    };
181    let target = {
182        let mut table = lock(&shared.table);
183        let found = if packet.your_discriminator != 0 {
184            match table.by_discr.get(&packet.your_discriminator) {
185                Some(e) if e.peer == peer => Some((e.inputs.clone(), e.conn)),
186                Some(_) => {
187                    tracing::warn!(
188                        "BFD dropped a packet from {from} naming another peer's session"
189                    );
190                    return None;
191                }
192                None => {
193                    tracing::debug!(
194                        "BFD dropped a packet from {from} for unknown discriminator {}",
195                        packet.your_discriminator
196                    );
197                    return None;
198                }
199            }
200        } else {
201            table
202                .by_peer
203                .get(&peer)
204                .and_then(|d| table.by_discr.get(d))
205                .map(|e| (e.inputs.clone(), e.conn))
206        };
207        match found {
208            Some(t) => t,
209            None => {
210                let now = Instant::now();
211                table.declined.retain(|_, until| *until > now);
212                if table.declined.contains_key(&peer) || table.asking.contains(&peer) {
213                    return None;
214                }
215                // The key is checked before the model hears of the peer: one that cannot
216                // authenticate is not a session request.
217                if let Err(e) = packet::verify(bytes, &packet, shared.auth.as_ref()) {
218                    tracing::warn!("BFD dropped a packet from {from} decision=dropped_auth: {e:#}");
219                    return None;
220                }
221                if table.by_discr.len() + table.asking.len() >= shared.max_sessions {
222                    tracing::warn!(
223                        "BFD ignored {from}: {} sessions already decision=max_sessions",
224                        shared.max_sessions
225                    );
226                    return None;
227                }
228                table.asking.insert(peer);
229                drop(table);
230                let (shared, bytes) = (shared.clone(), bytes.to_vec());
231                let handle = tokio::spawn(async move { ask(shared, peer, bytes, packet).await });
232                return Some(handle);
233            }
234        }
235    };
236    if target.0.try_send(input()).is_err() {
237        tracing::warn!("BFD dropped a packet from {from}: its session is {SESSION_INPUTS} behind");
238    }
239    shared
240        .ctx
241        .state
242        .update_connection_stats(
243            shared.ctx.server_id,
244            target.1,
245            Some(bytes.len() as u64),
246            None,
247            Some(1),
248            None,
249        )
250        .await;
251    None
252}
253
254/// A new peer: ask the model whether to bring a session up with it.
255async fn ask(shared: Shared, peer: IpAddr, bytes: Vec<u8>, packet: ControlPacket) {
256    let ctx = &shared.ctx;
257    let event = Event::new(
258        &actions::SESSION_REQUEST_EVENT,
259        json!({
260            "peer": peer.to_string(),
261            "peer_discriminator": packet.my_discriminator,
262            "peer_state": packet.state.name(),
263            "desired_min_tx_ms": packet.desired_min_tx_us / 1000,
264            "required_min_rx_ms": packet.required_min_rx_us / 1000,
265            "detect_mult": packet.detect_mult,
266            "authentication": packet.auth.as_ref().map(|a| a.kind.name()),
267            "multihop": shared.multihop,
268        }),
269    );
270    let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
271    let summary = format!("BFD session request from {peer}");
272    let log = Log::new(Some(&ctx.status_tx));
273    let accepted = match call_llm(
274        &ctx.llm_client,
275        &ctx.state,
276        ctx.server_id,
277        Some(id),
278        &event,
279        &BfdProtocol,
280    )
281    .await
282    {
283        Ok(execution) => execution.protocol_results.iter().find_map(|r| match r {
284            ActionResult::Custom { name, data } if name == actions::ACCEPT => Some(data.clone()),
285            _ => None,
286        }),
287        Err(e) => {
288            log.error(format!(
289                "{summary} decision=fail_closed_llm_error (no session: BFD has no refusal): {e}"
290            ));
291            None
292        }
293    };
294    let Some(answer) = accepted else {
295        log.info(format!(
296            "{summary} decision=model_silent (not accepted; ignored for {DECLINE_HOLD:?})"
297        ));
298        let mut table = lock(&shared.table);
299        table.asking.remove(&peer);
300        table.declined.insert(peer, Instant::now() + DECLINE_HOLD);
301        return;
302    };
303    let timers = match Timers::from_json(&answer, shared.timers) {
304        Ok(t) => t,
305        Err(e) => {
306            log.warn(format!(
307                "{summary} decision=fail_closed_action_error: {e:#}"
308            ));
309            lock(&shared.table).asking.remove(&peer);
310            return;
311        }
312    };
313    start_session(shared, peer, id, timers, bytes, packet).await;
314}
315
316async fn start_session(
317    shared: Shared,
318    peer: IpAddr,
319    id: ConnectionId,
320    timers: Timers,
321    bytes: Vec<u8>,
322    packet: ControlPacket,
323) {
324    let ctx = shared.ctx.clone();
325    let (inputs, inputs_rx) = mpsc::channel(SESSION_INPUTS);
326    let (notes_tx, notes) = mpsc::channel(NOTES);
327    let discr = {
328        let mut table = lock(&shared.table);
329        table.asking.remove(&peer);
330        let discr = table.new_discriminator();
331        table.by_discr.insert(
332            discr,
333            Entry {
334                peer,
335                inputs: inputs.clone(),
336                conn: id,
337            },
338        );
339        table.by_peer.insert(peer, discr);
340        discr
341    };
342    let remote = SocketAddr::new(peer, shared.port);
343    record(&ctx, id, remote, shared.tx.local_addr().ok()).await;
344    let peer_rx =
345        crate::server::peer_support::register_peer_channel(&ctx.state, ctx.server_id, id.as_u32())
346            .await;
347    Log::new(Some(&ctx.status_tx)).info(format!(
348        "BFD session {discr} with {peer} accepted decision=model_answered (tx {} ms, rx {} ms, x{})",
349        timers.desired_min_tx_us / 1000,
350        timers.required_min_rx_us / 1000,
351        timers.detect_mult
352    ));
353    let session = Session::new(discr, timers, shared.auth.clone(), true);
354    let _ = inputs.try_send(Input::Packet { bytes, packet });
355    let link = Link {
356        socket: shared.tx.clone(),
357        dest: remote,
358        label: format!("BFD session {discr} with {peer}"),
359    };
360    let run = tokio::spawn(runner::run(
361        session,
362        link,
363        inputs_rx,
364        notes_tx,
365        Some(SESSION_IDLE),
366    ));
367    ctx.state.register_server_task(ctx.server_id, run).await;
368    let watch = tokio::spawn(watch(shared, peer, discr, id, inputs, notes, peer_rx));
369    ctx.state.register_server_task(ctx.server_id, watch).await;
370}
371
372async fn record(
373    ctx: &SpawnContext,
374    id: ConnectionId,
375    remote: SocketAddr,
376    local: Option<SocketAddr>,
377) {
378    use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
379    let now = crate::utils::clock::Instant::now();
380    ctx.state
381        .add_connection_to_server(
382            ctx.server_id,
383            ConnectionState {
384                id,
385                remote_addr: remote,
386                local_addr: local.unwrap_or(remote),
387                bytes_sent: 0,
388                bytes_received: 0,
389                packets_sent: 0,
390                packets_received: 0,
391                last_activity: now,
392                status: ConnectionStatus::Active,
393                status_changed_at: now,
394                protocol_info: ProtocolConnectionInfo::new(json!({"state": "Down"})),
395            },
396        )
397        .await;
398}
399
400/// One session's model side: state changes become events, answers become session actions,
401/// and the dashboard's actions for this peer are forwarded in.
402async fn watch(
403    shared: Shared,
404    peer: IpAddr,
405    discr: u32,
406    id: ConnectionId,
407    inputs: mpsc::Sender<Input>,
408    mut notes: mpsc::Receiver<Note>,
409    mut peer_rx: mpsc::Receiver<crate::state::client_handles::ClientCommand>,
410) {
411    let ctx = &shared.ctx;
412    loop {
413        tokio::select! {
414            c = peer_rx.recv() => {
415                if let Some(command) = c {
416                    let action = command.action.clone();
417                    let _ = inputs.send(Input::Action { action, depth: 0, command: Some(command) }).await;
418                }
419            }
420            note = notes.recv() => match note {
421                None | Some(Note::Ended { .. }) => break,
422                Some(Note::Changed { previous, snapshot, depth }) => {
423                    let data = runner::state_event(&peer.to_string(), previous, &snapshot);
424                    Log::new(Some(&ctx.status_tx)).info(format!(
425                        "BFD session {discr} with {peer}: {} -> {} ({})",
426                        previous.name(), data["state"].as_str().unwrap_or_default(), data["diag"].as_str().unwrap_or_default()
427                    ));
428                    if depth >= MAX_FOLLOWUP_DEPTH {
429                        tracing::warn!("BFD session {discr}: not asking the model about this change, {depth} turns deep decision=followup_depth");
430                        continue;
431                    }
432                    for action in react(ctx, id, data).await {
433                        let _ = inputs.send(Input::Action { action, depth: depth + 1, command: None }).await;
434                    }
435                }
436            },
437        }
438    }
439    {
440        let mut table = lock(&shared.table);
441        table.by_discr.remove(&discr);
442        if table.by_peer.get(&peer) == Some(&discr) {
443            table.by_peer.remove(&peer);
444        }
445    }
446    ctx.state
447        .close_connection_on_server(ctx.server_id, id)
448        .await;
449}
450
451async fn react(ctx: &SpawnContext, id: ConnectionId, data: Value) -> Vec<Value> {
452    let event = Event::new(&actions::STATE_EVENT, data);
453    match call_llm(
454        &ctx.llm_client,
455        &ctx.state,
456        ctx.server_id,
457        Some(id),
458        &event,
459        &BfdProtocol,
460    )
461    .await
462    {
463        Ok(execution) => execution
464            .protocol_results
465            .iter()
466            .filter_map(|r| match r {
467                ActionResult::Custom { name, data } if name != actions::ACCEPT => {
468                    Some(data.clone())
469                }
470                _ => None,
471            })
472            .collect(),
473        Err(e) => {
474            // Nothing to fail closed to: the session keeps running as it is.
475            tracing::warn!("BFD state change not answered decision=fail_closed_llm_error: {e}");
476            vec![]
477        }
478    }
479}
netget/server/bfd/mod.rs

netget/server/bfd/
mod.rs
