netget/client/bfd/
mod.rs

1//! BFD client: one session with one router, in the active role, run by the server's engine
2//! (`src/server/bfd/runner.rs`). The model is asked when the session starts and whenever its
3//! state changes; its answers (timers, AdminDown, back up, disconnect) go to the session, as
4//! do actions injected through the client's command channel.
5pub mod actions;
6
7use crate::client::llm_budget::call_llm_for_client;
8use crate::llm::actions::client_trait::{Client, ClientActionResult};
9use crate::logging::emit::Log;
10use crate::protocol::{ConnectContext, Event};
11use crate::server::bfd::packet;
12use crate::server::bfd::runner::{self, Input, Link, Note};
13use crate::server::bfd::session::Session;
14use crate::server::bfd::ttl;
15use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
16use crate::state::{AccessLogOwner, ClientStatus};
17pub use actions::BfdClientProtocol;
18use anyhow::{Context, Result};
19use serde_json::{json, Value};
20use std::net::{IpAddr, SocketAddr};
21use std::sync::Arc;
22use tokio::sync::mpsc;
23
24/// How many model turns may follow from one another before the chain stops.
25pub const MAX_FOLLOWUP_DEPTH: u32 = 8;
26const TURNS: usize = 64;
27
28pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
29    let params = ctx.startup_params.as_ref();
30    let multihop_param = params
31        .map(|p| p.get_optional_bool("multihop"))
32        .transpose()?
33        .flatten();
34    let remote = parse_remote(&ctx.remote_addr, multihop_param).await?;
35    let multihop = multihop_param.unwrap_or(remote.port() == packet::MULTIHOP_PORT);
36    let local_ip = match params
37        .map(|p| p.get_optional_string("local_address"))
38        .transpose()?
39        .flatten()
40    {
41        Some(raw) => raw
42            .parse::<IpAddr>()
43            .with_context(|| format!("local_address {raw:?} is not an IP address"))?,
44        None => route_source(remote).await?,
45    };
46    let timers = crate::server::bfd::actions::timers_from(params)?;
47    let auth = crate::server::bfd::actions::auth_from(params)?;
48
49    // The router sends to its neighbour's address at the BFD port, so that is where to listen;
50    // SO_REUSEADDR lets a routing daemon's wildcard socket on the same port coexist.
51    let rx = crate::server::socket_helpers::create_reusable_udp_socket(SocketAddr::new(
52        local_ip,
53        remote.port(),
54    ))
55    .await
56    .with_context(|| format!("cannot listen on {local_ip}:{}", remote.port()))?;
57    ttl::enable(&rx).context("asking for the TTL of received packets")?;
58    let tx = Arc::new(runner::tx_socket(local_ip).await?);
59    let local = tx.local_addr()?;
60    let discr = loop {
61        let d: u32 = rand::random();
62        if d != 0 {
63            break d;
64        }
65    };
66    ctx.state
67        .update_client_status(ctx.client_id, ClientStatus::Connected)
68        .await;
69    Log::new(Some(&ctx.status_tx)).info(format!(
70        "BFD client {} session {discr} with {remote} from {local} ({})",
71        ctx.client_id,
72        if multihop { "multihop" } else { "single-hop" }
73    ));
74    let external =
75        crate::client::command_support::register_command_channel(&ctx.state, ctx.client_id).await;
76
77    let (inputs, inputs_rx) = mpsc::channel(64);
78    let (notes_tx, notes) = mpsc::channel(64);
79    let link = Link {
80        socket: tx,
81        dest: remote,
82        label: format!("BFD client {} session {discr}", ctx.client_id),
83    };
84    let session = Session::new(discr, timers, auth, false);
85    let run = tokio::spawn(runner::run(session, link, inputs_rx, notes_tx, None));
86    let run_abort = run.abort_handle();
87    ctx.state.register_client_task(ctx.client_id, run).await;
88
89    let reader_inputs = inputs.clone();
90    let reader = tokio::spawn(async move {
91        let mut buf = [0u8; 2048];
92        while let Ok((n, from, ttl)) = ttl::recv(&rx, &mut buf).await {
93            if from.ip() != remote.ip() {
94                continue;
95            }
96            if !multihop && ttl.is_some_and(|t| t != 255) {
97                tracing::warn!(
98                    "BFD client dropped a packet from {from} with TTL {ttl:?} decision=dropped_ttl"
99                );
100                continue;
101            }
102            let packet = match packet::decode(&buf[..n]) {
103                Ok(p) => p,
104                Err(e) => {
105                    tracing::warn!("BFD client dropped a datagram from {from}: {e:#}");
106                    continue;
107                }
108            };
109            if packet.your_discriminator != 0 && packet.your_discriminator != discr {
110                continue;
111            }
112            let input = Input::Packet {
113                bytes: buf[..n].to_vec(),
114                packet,
115            };
116            if reader_inputs.send(input).await.is_err() {
117                return;
118            }
119        }
120    });
121    let reader_abort = reader.abort_handle();
122    ctx.state.register_client_task(ctx.client_id, reader).await;
123
124    let (turn_tx, turns) = mpsc::channel::<(Event, u32)>(TURNS);
125    let (answers_tx, answers) = mpsc::channel::<(Value, u32)>(TURNS);
126    let _ = turn_tx.try_send((
127        Event::new(
128            &actions::STARTED_EVENT,
129            json!({"peer": remote.to_string(), "local": SocketAddr::new(local_ip, remote.port()).to_string(),
130                   "local_discriminator": discr, "multihop": multihop}),
131        ),
132        0,
133    ));
134    let dispatcher = tokio::spawn(run_turns(ctx.clone(), turns, answers_tx));
135    let dispatcher_abort = dispatcher.abort_handle();
136    ctx.state
137        .register_client_task(ctx.client_id, dispatcher)
138        .await;
139
140    let control_ctx = ctx.clone();
141    let peer = remote.ip().to_string();
142    let control = tokio::spawn(async move {
143        control(
144            &control_ctx,
145            &peer,
146            &inputs,
147            notes,
148            turn_tx,
149            answers,
150            external,
151        )
152        .await;
153        // Tell the router before going quiet (RFC 5880 §6.8.16): AdminDown, then stop.
154        let (reply_tx, reply) = tokio::sync::oneshot::channel();
155        let goodbye = Input::Action {
156            action: json!({"type": "bfd_admin_down", "diag": "administratively_down"}),
157            depth: MAX_FOLLOWUP_DEPTH,
158            command: Some(ClientCommand {
159                action: Value::Null,
160                reply_tx,
161            }),
162        };
163        if inputs.send(goodbye).await.is_ok() {
164            let _ = tokio::time::timeout(std::time::Duration::from_secs(1), reply).await;
165        }
166        run_abort.abort();
167        reader_abort.abort();
168        dispatcher_abort.abort();
169        control_ctx
170            .state
171            .update_client_status(control_ctx.client_id, ClientStatus::Disconnected)
172            .await;
173        control_ctx
174            .state
175            .remove_client_handle(control_ctx.client_id)
176            .await;
177        let _ = control_ctx.status_tx.send("__UPDATE_UI__".into());
178    });
179    ctx.state.register_client_task(ctx.client_id, control).await;
180    Ok(local)
181}
182
183/// `ip` or `ip:port`; the port defaults to the BFD port for the mode.
184async fn parse_remote(raw: &str, multihop: Option<bool>) -> Result<SocketAddr> {
185    let raw = raw.trim();
186    anyhow::ensure!(!raw.is_empty(), "remote_addr is the router's address");
187    if let Ok(ip) = raw.parse::<IpAddr>() {
188        let port = if multihop == Some(true) {
189            packet::MULTIHOP_PORT
190        } else {
191            packet::SINGLE_HOP_PORT
192        };
193        return Ok(SocketAddr::new(ip, port));
194    }
195    tokio::net::lookup_host(raw)
196        .await
197        .with_context(|| format!("cannot resolve {raw}"))?
198        .next()
199        .with_context(|| format!("{raw} resolved to no address"))
200}
201
202/// The source address the route to `remote` uses.
203async fn route_source(remote: SocketAddr) -> Result<IpAddr> {
204    let probe = tokio::net::UdpSocket::bind(if remote.is_ipv4() {
205        "0.0.0.0:0"
206    } else {
207        "[::]:0"
208    })
209    .await?;
210    probe
211        .connect(remote)
212        .await
213        .with_context(|| format!("no route to {remote}"))?;
214    Ok(probe.local_addr()?.ip())
215}
216
217async fn control(
218    ctx: &ConnectContext,
219    peer: &str,
220    inputs: &mpsc::Sender<Input>,
221    mut notes: mpsc::Receiver<Note>,
222    turns: mpsc::Sender<(Event, u32)>,
223    mut answers: mpsc::Receiver<(Value, u32)>,
224    mut external: mpsc::Receiver<ClientCommand>,
225) {
226    loop {
227        tokio::select! {
228            c = external.recv() => {
229                let Some(command) = c else { return };
230                ctx.state
231                    .record_access_log(
232                        AccessLogOwner::Client(ctx.client_id.as_u32()),
233                        "BFD",
234                        None,
235                        "injected_action",
236                        command.action.clone(),
237                        vec![],
238                    )
239                    .await;
240                match BfdClientProtocol.execute_action(command.action.clone()) {
241                    Ok(ClientActionResult::Disconnect) => {
242                        crate::client::command_support::reply(command, Ok(ClientSendOutcome::Disconnected));
243                        return;
244                    }
245                    Ok(_) => {
246                        let action = command.action.clone();
247                        let _ = inputs.send(Input::Action { action, depth: 0, command: Some(command) }).await;
248                    }
249                    Err(e) => crate::client::command_support::reply(
250                        command,
251                        Ok(ClientSendOutcome::Rejected { error: e.to_string() }),
252                    ),
253                }
254            }
255            a = answers.recv() => {
256                let Some((action, depth)) = a else { return };
257                match BfdClientProtocol.execute_action(action.clone()) {
258                    Ok(ClientActionResult::Disconnect) => return,
259                    Ok(_) => {
260                        let _ = inputs.send(Input::Action { action, depth, command: None }).await;
261                    }
262                    Err(e) => tracing::warn!("BFD client {} refused {}: {e:#}", ctx.client_id, action["type"]),
263                }
264            }
265            n = notes.recv() => match n {
266                None | Some(Note::Ended { .. }) => return,
267                Some(Note::Changed { previous, snapshot, depth }) => {
268                    let data = runner::state_event(peer, previous, &snapshot);
269                    Log::new(Some(&ctx.status_tx)).info(format!(
270                        "BFD client {}: {} -> {} ({})",
271                        ctx.client_id,
272                        previous.name(),
273                        data["state"].as_str().unwrap_or_default(),
274                        data["diag"].as_str().unwrap_or_default()
275                    ));
276                    if turns.try_send((Event::new(&actions::STATE_EVENT, data), depth)).is_err() {
277                        tracing::warn!("BFD client {} dropped a state event: the model is {TURNS} behind decision=turn_queue_full", ctx.client_id);
278                    }
279                }
280            },
281        }
282    }
283}
284
285async fn run_turns(
286    ctx: ConnectContext,
287    mut turns: mpsc::Receiver<(Event, u32)>,
288    answers: mpsc::Sender<(Value, u32)>,
289) {
290    let protocol = BfdClientProtocol;
291    while let Some((event, depth)) = turns.recv().await {
292        if depth >= MAX_FOLLOWUP_DEPTH {
293            tracing::warn!("BFD client {} not asking the model about {}: {depth} turns deep decision=followup_depth", ctx.client_id, event.id());
294            continue;
295        }
296        let instruction = ctx
297            .state
298            .get_instruction_for_client(ctx.client_id)
299            .await
300            .unwrap_or_default();
301        let memory = ctx
302            .state
303            .get_memory_for_client(ctx.client_id)
304            .await
305            .unwrap_or_default();
306        match call_llm_for_client(
307            &ctx.llm_client,
308            &ctx.state,
309            ctx.client_id.to_string(),
310            &instruction,
311            &memory,
312            Some(&event),
313            &protocol,
314            &ctx.status_tx,
315        )
316        .await
317        {
318            Ok(result) => {
319                if let Some(memory) = result.memory_updates {
320                    ctx.state.set_memory_for_client(ctx.client_id, memory).await;
321                }
322                for action in result.actions {
323                    if answers.send((action, depth + 1)).await.is_err() {
324                        return;
325                    }
326                }
327            }
328            Err(e) => Log::new(Some(&ctx.status_tx)).warn(format!("BFD client handler: {e}")),
329        }
330    }
331}
netget/client/bfd/mod.rs

netget/client/bfd/
mod.rs
