From bef875e50bdf33c1249e407cae0424b9beeb818e Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Sat, 10 Oct 2026 22:44:32 +0200 Subject: [PATCH] session: the way out to the peer as a type of its own A code-quality pass over the per-peer session. The outgoing side of a session, eight fields served from six places in a two-thousand-line file, is `Outgoing`: the write buffer, the batches waiting for the writer task, the task and its word on why it ended. One real race went with it: when the writer task failed with the queue full, the session took whichever of two ready branches the select chose, and half the time ended with an invented send hold timer expiry instead of the connection's error, which decides whether the peer's routes are kept over a restart; the writer's own word is taken now. The run loop's handling of what the rest of the server has to send is a method that goes by the state machine's state, so that pieces after a NOTIFICATION in the same batch do not follow it out; `execute_all` reads one line an action with the sending, delivering and after-failure steps named. A peer's routes have one way onto the table at a session's end, whether removed or kept for a restart. An UPDATE whose attributes cannot be kept ends the session under an error of its own, `Unstorable`, where a message of ours that cannot be encoded is dropped. What is kept of a restarting peer is a value, not seven arguments. The dialer no longer turns a session's end into a Result and back. The server's screening of a connection is called that, apart from the collision check's `admit`. Testing: the back-pressure machinery has unit tests at last: batches wait while the writer has no room, room goes to the oldest batch, a piece is counted as the messages in it, and a lost connection is reported as one while batches wait. The routes module gains tests of holding at its bound, forgetting and revalidation. The server tests and the whole integration suite pass. Co-Authored-By: Claude Fable 5.1 --- crates/server/src/connection.rs | 798 +++++++++++++++++++++----------- crates/server/src/dialer.rs | 30 +- crates/server/src/error.rs | 10 +- crates/server/src/restart.rs | 57 +-- crates/server/src/routes.rs | 261 ++++++++++- crates/server/src/server.rs | 9 +- 6 files changed, 842 insertions(+), 323 deletions(-) diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index ee98e80..d7aeae5 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -28,9 +28,9 @@ use bgp_rib::table::{Outcome, PeerIndex, PeerInfo, RibTable}; use bgp_rpki::Rpki; use bytes::BytesMut; use prometheus::IntGauge; -use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::net::TcpStream; -use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::net::tcp::OwnedReadHalf; use tokio::sync::mpsc::error::TrySendError; use tokio::sync::{mpsc, watch}; use tokio::task::JoinHandle; @@ -42,7 +42,7 @@ use crate::collision::{CollisionTable, ConnectionGuard, Direction}; use crate::config::PeerConfig; use crate::error::SessionError; use crate::family::FamilySet; -use crate::fsm::{Action, Event, Fsm, FsmConfig, Negotiated, Received}; +use crate::fsm::{Action, Event, Fsm, FsmConfig, Negotiated, Received, State}; use crate::holddown::{HeldDown, HoldDowns}; use crate::metrics::{ EXPORT_LIMIT_EXCEEDED, MESSAGES_RECEIVED, MESSAGES_SENT, PEER_ADVERTISED, PEER_ROUTES, @@ -52,7 +52,7 @@ use crate::panic::contained; use crate::policy::{self, ExportLimits, ExportPolicy, ImportPolicy, PeerRole}; use crate::reach::{Reach, neighbor_at}; use crate::rejections::{PeerRejections, RejectionLog}; -use crate::restart::{self, Agreed, GracefulRestartConfig, Leave, Restarts}; +use crate::restart::{self, Agreed, GracefulRestartConfig, Kept, Leave, Restarts}; use crate::routes::{self, Held}; use crate::rpki::{Checks, Subject, Validator}; use crate::session::{ @@ -61,8 +61,8 @@ use crate::session::{ use crate::throttle::Throttle; /// How many items may be queued for a session before senders wait: each -/// a message, or a piece of up to [`export::ENCODED_PIECE`] octets of -/// UPDATEs from the exporter, so a few megabytes of a table at most. +/// a message, or a piece of up to [`crate::export::ENCODED_PIECE`] octets +/// of UPDATEs from the exporter, so a few megabytes of a table at most. const OUTBOUND_QUEUE_CAPACITY: usize = 64; /// How many octets of messages are gathered before they are written to @@ -72,8 +72,10 @@ const OUTBOUND_QUEUE_CAPACITY: usize = 64; /// round trip. Gathered, they go as full segments in one write. const WRITE_BATCH_BYTES: usize = 64 * 1024; -/// How many batches may wait for the writer task at a time: a megabyte -/// or so a session, beyond what the socket's own buffer takes. +/// How many batches may wait for the writer task at a time. A batch is +/// a batch's worth and at most one piece from the exporter more, so this +/// is two megabytes or so a session, beyond what the socket's own buffer +/// takes. const WRITER_QUEUE: usize = 16; /// The least time a write to the peer may make no progress before the @@ -205,7 +207,7 @@ pub struct SessionEnd { } /// Why a session ended when nothing went wrong. -pub const CLOSED_AT_OUR_REQUEST: &str = "session closed at our request"; +const CLOSED_AT_OUR_REQUEST: &str = "session closed at our request"; impl SessionEnd { /// Why the session ended, in a sentence. @@ -342,11 +344,13 @@ async fn release_routes( let kept = routes.keep(versions.clone()).await; session.server.restarts.keep( (session.peer_addr.into(), peer_id.asn), - session.peer_addr, - session.peer.name.clone(), - rib_peer, - versions, - kept, + Kept::new( + session.peer_addr, + session.peer.name.clone(), + rib_peer, + versions, + kept, + ), time, ); info!( @@ -373,7 +377,8 @@ struct Established { peer_id: PeerId, /// The peer's number in the routing table. rib_peer: PeerIndex, - /// Messages the rest of the server queued for the peer. + /// What the rest of the server queued for the peer: messages, and + /// pieces of UPDATEs the exporter encoded. outbound: mpsc::Receiver, /// Where the rest of the server asks for the session to end, and why. ending: mpsc::UnboundedReceiver, @@ -497,28 +502,21 @@ impl PeerRoutes { } /// Removes the routes and says what that did. - async fn remove(mut self) -> Outcome { + async fn remove(self) -> Outcome { let peer = self.peer; - match off_thread(&self.table, move |table| table.remove_peer(peer)).await { - Ok(outcome) => { - self.removed = true; - outcome - } - // Not removed, then, or not all of them: dropping this tries - // once more. - Err(e) => { - error!(err = %e, "Removing a peer's routes failed; trying again"); - Outcome::default() - } - } + self.on_table("Removing a peer's routes", move |table| { + table.remove_peer(peer) + }) + .await + .unwrap_or_default() } /// Keeps the routes of the IP versions `versions` for the peer's next /// session, marked stale, and removes the others. Returns how many /// were kept. - async fn keep(mut self, versions: Vec) -> usize { - let (table, peer) = (Arc::clone(&self.table), self.peer); - let kept = tokio::task::spawn_blocking(move || { + async fn keep(self, versions: Vec) -> usize { + let peer = self.peer; + self.on_table("Keeping a restarting peer's routes", move |table| { let all = [IpVersion::Ipv4, IpVersion::Ipv6]; let marked = table.mark_stale(peer, &all); let others: Vec = all @@ -528,16 +526,27 @@ impl PeerRoutes { let removed = table.sweep_stale(peer, &others).removed; marked.saturating_sub(removed) }) - .await; - match kept { - Ok(kept) => { + .await + .unwrap_or(0) + } + + /// Does `work` to the routes on a thread that can take its time, and + /// takes that as done with them. If it fails, which is only if it + /// panicked, `what` is logged and the routes are all removed when + /// this is dropped. + async fn on_table( + mut self, + what: &'static str, + work: impl FnOnce(&RibTable) -> T + Send + 'static, + ) -> Option { + match off_thread(&self.table, work).await { + Ok(outcome) => { self.removed = true; - kept + Some(outcome) } - // Dropping this removes them all. Err(e) => { - error!(err = %e, "Keeping a restarting peer's routes failed; removing them"); - 0 + error!(err = %e, "{what} failed; the routes are removed instead"); + None } } } @@ -613,8 +622,8 @@ enum Wake { Event(Event), /// The writer has room for a batch that was waiting. WriterRoom(mpsc::OwnedPermit), - /// Messages from the rest of the server to send, as many as were - /// waiting, to be written together. + /// What the rest of the server has to send, messages and pieces of + /// encoded UPDATEs, as many as were waiting, to be written together. Outbound(Vec), /// The validation data has changed; these routes of the peer's are /// now invalid. @@ -649,33 +658,14 @@ struct Session { said: Said, fsm: Fsm, frames: FrameReader, - /// Batches of messages on their way to the writer task, which writes - /// them to the peer; see [`write_to_peer`]. - writer: mpsc::Sender, - writer_task: JoinHandle<()>, - /// Batches the writer had no room for yet, oldest first. While there - /// are any, no more messages are taken from the rest of the server; - /// the peer is still read. - pending: VecDeque, - /// What the writer task says if it fails. - write_failed: mpsc::Receiver, - /// Messages encoded and not yet made a batch of. - write_buf: BytesMut, + /// Messages on their way to the peer. + outgoing: Outgoing, /// How messages from the peer are decoded: the width of AS numbers and /// where prefixes carry path identifiers are known once the OPENs are /// exchanged. ctx: ParserContext, - /// How messages to the peer are encoded. Its own context: whether path - /// identifiers are sent is settled apart from whether they are - /// received (RFC 7911 §4). - send_ctx: ParserContext, hold_timer: Option>>, keepalive_timer: Option, - /// How long a write may make no progress; see [`MIN_SEND_HOLD_TIME`]. - /// The writer task watches it. - send_hold: watch::Sender, - /// Whether a NOTIFICATION of ours is on its way to the peer. - notified: bool, /// This connection's slot in the collision table, once admitted. collision: Option, established: Option, @@ -707,10 +697,6 @@ impl Session { // wait for the acknowledgement of the one before it. let _ = stream.set_nodelay(true); let (reader, writer) = stream.into_split(); - let (batches, queue) = mpsc::channel(WRITER_QUEUE); - let (failed, write_failed) = mpsc::channel(1); - let (send_hold, hold) = watch::channel(MIN_SEND_HOLD_TIME); - let writer_task = tokio::spawn(write_to_peer(writer, queue, hold, failed)); let (fsm, first_actions) = Fsm::new(FsmConfig { router_id: server.router_id, local_asn: server.local_asn, @@ -735,17 +721,10 @@ impl Session { rejections: server.rejections.of_peer(peer_addr.into(), &label), fsm, frames: FrameReader::new(reader), - writer: batches, - writer_task, - pending: VecDeque::new(), - write_failed, - write_buf: BytesMut::new(), + outgoing: Outgoing::new(writer), ctx: ParserContext::default(), - send_ctx: ParserContext::default(), hold_timer: None, keepalive_timer: None, - send_hold, - notified: false, collision: None, established: None, peer_addr, @@ -773,15 +752,14 @@ impl Session { // Whatever was gathered to be written when the panic came is // not to go; the NOTIFICATION is, now. Contained as well: // whatever panicked may do so again. - self.write_buf.clear(); - self.pending.clear(); + self.outgoing.discard(); if !matches!( contained(async { self.send(¬ification) }).await, Ok(Ok(())) ) { warn!(peer_addr = %socket_text(self.peer_addr), "Could not send NOTIFICATION after a panic"); } - self.flush(); + self.outgoing.flush(); } error } @@ -791,21 +769,9 @@ impl Session { async fn close(mut self) { // A new connection from the peer must not find this one in its way. self.collision = None; - // Whatever is still to go out, a NOTIFICATION among it, is given to - // the writer, which closes our side once it has written it. - self.flush(); - let hand_over = async { - for batch in std::mem::take(&mut self.pending) { - if self.writer.send(batch).await.is_err() { - break; - } - } - }; - let _ = timeout(CLOSE_LINGER, hand_over).await; - let (closed, _) = mpsc::channel(1); - drop(std::mem::replace(&mut self.writer, closed)); - let _ = timeout(CLOSE_LINGER, &mut self.writer_task).await; - if !self.notified { + let notified = self.outgoing.notified; + self.outgoing.close().await; + if !notified { return; } // Our side is done sending; what the peer still sends is read and @@ -821,30 +787,9 @@ impl Session { } actions = match self.next_event().await? { Wake::Event(event) => self.fsm.handle(event), - Wake::Outbound(messages) => { - let mut actions = Vec::new(); - for item in messages { - match item { - Outbound::Message(msg) => { - actions.extend(self.fsm.handle(Event::Outbound(msg))); - } - // Encoded by the exporter for this session as - // it is now; one for a session that is no - // longer established is void. - Outbound::Encoded(update) => { - if self.established.is_some() { - self.send_encoded(update); - } - } - } - } - actions - } + Wake::Outbound(items) => self.take_outbound(items), Wake::WriterRoom(permit) => { - if let Some(batch) = self.pending.pop_front() { - permit.send(batch); - } - self.push_pending(); + self.outgoing.use_room(permit); vec![] } Wake::Revalidate(suspects) => self.revalidate(&suspects), @@ -860,6 +805,7 @@ impl Session { /// Waits for whatever happens next. Every branch is cancel-safe: the /// branches that lose leave nothing half done. async fn next_event(&mut self) -> Result { + let may_take_more = self.outgoing.may_take_more(); let event = tokio::select! { frame = self.frames.next(&self.ctx) => match frame { Ok(mut decoded) => { @@ -883,15 +829,11 @@ impl Session { }, () = expired(&mut self.hold_timer) => Event::HoldTimerExpired, () = ticked(&mut self.keepalive_timer) => Event::KeepaliveTimerExpired, - Some(error) = self.write_failed.recv() => Event::LocalFailure(error), - permit = self.writer.clone().reserve_owned(), if !self.pending.is_empty() => { - return Ok(match permit { - Ok(permit) => Wake::WriterRoom(permit), - // The writer task is gone; its word is on its way. - Err(_) => return Ok(Wake::Event(Event::LocalFailure(SessionError::SendHoldTimerExpired(*self.send_hold.borrow())))), - }); - } - wake = from_server(&mut self.established, self.pending.is_empty()) => return Ok(wake), + room = self.outgoing.writer_says() => match room { + Ok(permit) => return Ok(Wake::WriterRoom(permit)), + Err(error) => Event::LocalFailure(error), + }, + wake = from_server(&mut self.established, may_take_more) => return Ok(wake), () = killed(&mut self.collision) => Event::CollisionLost, leave = leaving(&mut self.leave) => Event::Leave(leave), }; @@ -928,7 +870,7 @@ impl Session { async fn execute(&mut self, actions: Vec) -> Option> { let result = self.execute_all(actions).await; // Whatever came of the actions goes to the writer as one batch. - self.flush(); + self.outgoing.flush(); result } @@ -936,26 +878,11 @@ impl Session { let mut queue: VecDeque = actions.into(); while let Some(action) = queue.pop_front() { match action { - Action::Send(msg) => match self.send(&msg) { - Ok(()) => {} - // The session is ending anyway and the Close that - // follows says why; a NOTIFICATION is best effort. - Err(e) if matches!(msg, Message::Notification(_)) => { - warn!(peer_addr = %socket_text(self.peer_addr), err = %e, "Failed to send NOTIFICATION"); + Action::Send(msg) => { + if let Err(e) = self.send_or_drop(&msg) { + return Some(Err(e)); } - // A message of our own making that cannot be written. - // That is a bug here, and nothing the peer did: the - // session is worth more than the message. - Err(SessionError::Encode(e)) => { - error!( - peer_addr = %socket_text(self.peer_addr), - message = message_type_name(&msg), - err = %e, - "Dropped a message that cannot be encoded" - ); - } - Err(e) => return Some(Err(e)), - }, + } Action::ResetHoldTimer(duration) => match &mut self.hold_timer { Some(timer) => timer.as_mut().reset(Instant::now() + duration), None => self.hold_timer = Some(Box::pin(sleep(duration))), @@ -970,17 +897,17 @@ impl Session { timer.set_missed_tick_behavior(MissedTickBehavior::Delay); self.keepalive_timer = Some(timer); } + // What the machine wanted next no longer applies once + // what it wanted now could not be done. Action::Admit(negotiated) => { if !self.admit(&negotiated) { - // What the machine wanted next no longer applies. queue = self.fsm.handle(Event::CollisionLost).into(); } } Action::Established(negotiated) => { if let Err(e) = self.establish(&negotiated) { - // What the machine wanted next no longer applies. error!(peer_addr = %socket_text(self.peer_addr), err = %e, "Cannot take the peer's routes"); - queue = self.fsm.handle(Event::LocalFailure(e)).into(); + queue = self.after_failure(e); } } Action::RoutesIgnored(count) => { @@ -997,21 +924,8 @@ impl Session { update, attributes_wire, } => { - if let Err(e) = self.store(&update, attributes_wire.as_deref()) { - match &e { - SessionError::PrefixLimitExceeded { .. } => warn!( - peer_addr = %socket_text(self.peer_addr), - err = %e, - hold_down = ?self.peer.import.hold_down, - "Ending the session and keeping the peer down" - ), - _ => error!( - peer_addr = %socket_text(self.peer_addr), - err = %e, - "Cannot store the peer's routes" - ), - } - queue = self.fsm.handle(Event::LocalFailure(e)).into(); + if let Err(e) = self.deliver(&update, attributes_wire.as_deref()) { + queue = self.after_failure(e); } } Action::EndOfRib(family) => self.end_of_rib(family), @@ -1033,6 +947,70 @@ impl Session { None } + /// Sends a message of ours, or drops it if it cannot be encoded: that + /// is a bug here, and nothing the peer did, and the session is worth + /// more than the message. A NOTIFICATION that cannot be sent is let + /// go of too: the session is ending anyway, and the Close that follows + /// says why. + fn send_or_drop(&mut self, msg: &Message) -> Result<(), SessionError> { + match self.send(msg) { + Ok(()) => Ok(()), + Err(e) if matches!(msg, Message::Notification(_)) => { + warn!(peer_addr = %socket_text(self.peer_addr), err = %e, "Failed to send NOTIFICATION"); + Ok(()) + } + Err(SessionError::Encode(e)) => { + error!( + peer_addr = %socket_text(self.peer_addr), + message = message_type_name(msg), + err = %e, + "Dropped a message that cannot be encoded" + ); + Ok(()) + } + Err(e) => Err(e), + } + } + + /// Applies a received UPDATE to the table, and says so if that fails. + fn deliver( + &mut self, + update: &UpdateMessage, + attributes_wire: Option<&[u8]>, + ) -> Result<(), SessionError> { + let outcome = self.apply_update(update, attributes_wire); + if let Err(e) = &outcome { + match e { + SessionError::PrefixLimitExceeded { .. } => warn!( + peer_addr = %socket_text(self.peer_addr), + err = %e, + hold_down = ?self.peer.import.hold_down, + "Ending the session and keeping the peer down" + ), + // This ends the session where a message of ours that + // cannot be encoded does not: the peer's routes would be + // wrong from here on, and a reset puts them right. + SessionError::Unstorable(_) => error!( + peer_addr = %socket_text(self.peer_addr), + err = %e, + "Cannot store the peer's routes" + ), + _ => error!( + peer_addr = %socket_text(self.peer_addr), + err = %e, + "Cannot apply the peer's UPDATE" + ), + } + } + outcome + } + + /// What the machine wants done after a failure here, in place of + /// whatever it wanted before. + fn after_failure(&mut self, error: SessionError) -> VecDeque { + self.fsm.handle(Event::LocalFailure(error)).into() + } + /// Takes what the OPENs settled into use, and asks collision detection /// whether this connection may go on. Returns whether it may. fn admit(&mut self, negotiated: &Negotiated) -> bool { @@ -1041,13 +1019,13 @@ impl Session { ibgp: negotiated.ibgp, add_path: negotiated.add_path_receive.iter().collect(), }; - self.send_ctx = ParserContext { - add_path: negotiated.add_path_send.iter().collect(), - ..self.ctx.clone() - }; - let hold_time = Duration::from_secs(negotiated.hold_time_secs.into()); - self.send_hold - .send_replace(MIN_SEND_HOLD_TIME.max(2 * hold_time)); + self.outgoing.negotiated( + ParserContext { + add_path: negotiated.add_path_send.iter().collect(), + ..self.ctx.clone() + }, + Duration::from_secs(negotiated.hold_time_secs.into()), + ); self.collision = self.server.collisions.admit( self.server.router_id, @@ -1329,7 +1307,7 @@ impl Session { } /// Puts the routes of a received UPDATE into the routing table. - fn store( + fn apply_update( &mut self, update: &UpdateMessage, attributes_wire: Option<&[u8]>, @@ -1391,8 +1369,8 @@ impl Session { vec![] } - /// Puts `msg` in the write buffer, and makes a batch of the buffer if - /// it has a batch's worth; the rest goes with [`Session::flush`]. + /// Puts `msg` on its way to the peer (see [`Outgoing::send`]), and says + /// so where it is worth saying. fn send(&mut self, msg: &Message) -> Result<(), SessionError> { if let (Message::Update(update), Some(established)) = (msg, &self.established) { let (withdrawn, announced) = prefix_counts(update); @@ -1400,67 +1378,28 @@ impl Session { } if let Message::Notification(notification) = msg { info!(peer_addr = %socket_text(self.peer_addr), error = %notification.error, "Sending NOTIFICATION, closing session"); - self.notified = true; } - let before = self.write_buf.len(); - if let Err(e) = msg.to_wire(&self.send_ctx, &mut self.write_buf) { - self.write_buf.truncate(before); - return Err(e.into()); - } - MESSAGES_SENT - .with_label_values(&[message_type_name(msg)]) - .inc(); - if self.write_buf.len() >= WRITE_BATCH_BYTES { - self.flush(); - } - Ok(()) + self.outgoing.send(msg) } - /// Puts a piece of UPDATEs that are already encoded in the write - /// buffer. - fn send_encoded(&mut self, piece: BytesMut) { - // Counted by their headers: a piece is of whole messages. - let mut messages = 0; - let mut at = 0; - while at + 18 <= piece.len() { - at += usize::from(u16::from_be_bytes([piece[at + 16], piece[at + 17]])).max(19); - messages += 1; - } - self.write_buf.extend_from_slice(&piece); - MESSAGES_SENT - .with_label_values(&["update"]) - .inc_by(messages); - if self.write_buf.len() >= WRITE_BATCH_BYTES { - self.flush(); - } - } - - /// Makes a batch of the write buffer and gives it to the writer, or - /// keeps it until the writer has room. Never waits: a session that - /// waited to write would not be reading, and a peer doing the same - /// would wait on it for ever. - fn flush(&mut self) { - if self.write_buf.is_empty() { - return; - } - // Split off, not copied: the buffer's bytes go as they are. - self.pending.push_back(self.write_buf.split()); - self.push_pending(); - } - - /// Gives the writer what it has room for, oldest first. - fn push_pending(&mut self) { - while let Some(batch) = self.pending.pop_front() { - match self.writer.try_send(batch) { - Ok(()) => {} - Err(TrySendError::Full(batch)) => { - self.pending.push_front(batch); - return; + /// Takes what the rest of the server has to send: a message goes + /// through the state machine, which says what to do with it; a piece + /// of UPDATEs the exporter encoded for this session as it is now goes + /// to the peer as it is, unless the session is no longer established, + /// as after a NOTIFICATION earlier in the same batch, when it is void. + fn take_outbound(&mut self, items: Vec) -> Vec { + let mut actions = Vec::new(); + for item in items { + match item { + Outbound::Message(msg) => actions.extend(self.fsm.handle(Event::Outbound(msg))), + Outbound::Encoded(piece) => { + if self.fsm.state() == State::Established { + self.outgoing.send_piece(piece); + } } - // The writer task is gone; its word is on its way. - Err(TrySendError::Closed(_)) => return, } } + actions } } @@ -1556,6 +1495,211 @@ async fn from_server(established: &mut Option, may_take_outbound: b } } +// --------------------------------------------------------------------------- +// The way out to the peer +// --------------------------------------------------------------------------- + +/// A session's messages on their way to the peer: gathered into batches, +/// handed to a task that writes them, and kept here while that task has +/// no room for them. +/// +/// Nothing here waits on the peer. A session that waited to write would +/// not be reading, and a peer doing the same would wait on it for ever. +/// What the writer has no room for is kept in `pending`, and while there +/// is any, nothing more is taken from the rest of the server; the peer is +/// still read. +struct Outgoing { + /// Batches of messages on their way to the writer task, which writes + /// them to the peer; see [`write_to_peer`]. + writer: mpsc::Sender, + writer_task: JoinHandle<()>, + /// Batches the writer had no room for yet, oldest first. + pending: VecDeque, + /// What the writer task says if it fails. + write_failed: mpsc::Receiver, + /// Messages encoded and not yet made a batch of. + write_buf: BytesMut, + /// How messages to the peer are encoded. Its own context, apart from + /// the one messages from the peer are decoded with: whether path + /// identifiers are sent is settled apart from whether they are + /// received (RFC 7911 §4). + send_ctx: ParserContext, + /// How long a write may make no progress; see [`MIN_SEND_HOLD_TIME`]. + /// The writer task watches it. + send_hold: watch::Sender, + /// Whether a NOTIFICATION of ours is on its way to the peer. + notified: bool, +} + +impl Outgoing { + /// The way out over `writer`, with nothing on it yet. + fn new(writer: impl AsyncWrite + Send + Unpin + 'static) -> Self { + let (batches, queue) = mpsc::channel(WRITER_QUEUE); + let (failed, write_failed) = mpsc::channel(1); + let (send_hold, hold) = watch::channel(MIN_SEND_HOLD_TIME); + Self { + writer: batches, + writer_task: tokio::spawn(write_to_peer(writer, queue, hold, failed)), + pending: VecDeque::new(), + write_failed, + write_buf: BytesMut::new(), + send_ctx: ParserContext::default(), + send_hold, + notified: false, + } + } + + /// Takes what the OPENs settled into use: how messages are encoded for + /// the peer, and the hold time, which sets how long a write may make + /// no progress. + fn negotiated(&mut self, send_ctx: ParserContext, hold_time: Duration) { + self.send_ctx = send_ctx; + self.send_hold + .send_replace(MIN_SEND_HOLD_TIME.max(2 * hold_time)); + } + + /// Whether more may be taken from the rest of the server: not while + /// the writer is behind with what it has. + fn may_take_more(&self) -> bool { + self.pending.is_empty() + } + + /// Puts `msg` in the write buffer, and makes a batch of the buffer if + /// it has a batch's worth; the rest goes with [`Outgoing::flush`]. + fn send(&mut self, msg: &Message) -> Result<(), SessionError> { + if let Message::Notification(_) = msg { + self.notified = true; + } + let before = self.write_buf.len(); + if let Err(e) = msg.to_wire(&self.send_ctx, &mut self.write_buf) { + self.write_buf.truncate(before); + return Err(e.into()); + } + MESSAGES_SENT + .with_label_values(&[message_type_name(msg)]) + .inc(); + if self.write_buf.len() >= WRITE_BATCH_BYTES { + self.flush(); + } + Ok(()) + } + + /// Puts a piece of UPDATEs that are already encoded in the write + /// buffer. + fn send_piece(&mut self, piece: BytesMut) { + // Counted by their headers: a piece is of whole messages. + let mut messages = 0; + let mut at = 0; + while let Ok(Some(length)) = frame_length(&piece[at..]) { + at += length; + messages += 1; + } + self.write_buf.extend_from_slice(&piece); + MESSAGES_SENT + .with_label_values(&["update"]) + .inc_by(messages); + if self.write_buf.len() >= WRITE_BATCH_BYTES { + self.flush(); + } + } + + /// Makes a batch of the write buffer and gives it to the writer, or + /// keeps it until the writer has room. Never waits. + fn flush(&mut self) { + if self.write_buf.is_empty() { + return; + } + // Split off, not copied: the buffer's bytes go as they are. + self.pending.push_back(self.write_buf.split()); + self.push_pending(); + } + + /// Gives the writer what it has room for, oldest first. + fn push_pending(&mut self) { + while let Some(batch) = self.pending.pop_front() { + match self.writer.try_send(batch) { + Ok(()) => {} + Err(TrySendError::Full(batch)) => { + self.pending.push_front(batch); + return; + } + // The writer task is gone; its word is on its way. + Err(TrySendError::Closed(_)) => return, + } + } + } + + /// Gives the writer the oldest batch that was waiting, in the room it + /// has made, and whatever else it has room for. + fn use_room(&mut self, permit: mpsc::OwnedPermit) { + if let Some(batch) = self.pending.pop_front() { + permit.send(batch); + } + self.push_pending(); + } + + /// Throws away whatever was gathered and not yet given to the writer. + fn discard(&mut self) { + self.write_buf.clear(); + self.pending.clear(); + } + + /// What the writer task has to say, once it has something: that it has + /// room for a batch that is waiting, if one is, or else why it failed. + /// Cancel-safe. + async fn writer_says(&mut self) -> Result, SessionError> { + if self.pending.is_empty() { + return Err(self.failed().await); + } + self.room().await + } + + /// Room in the writer's queue for one batch, once it has some. + async fn room(&mut self) -> Result, SessionError> { + match self.writer.clone().reserve_owned().await { + Ok(permit) => Ok(permit), + // The writer task is gone. It said why before it went, and + // that is the word to end the session with: the peer's routes + // are kept over a lost connection and not over a send hold + // timer. + Err(_) => Err(self.failed().await), + } + } + + /// Why the writer task failed, once it has. + async fn failed(&mut self) -> SessionError { + match self.write_failed.recv().await { + Some(error) => error, + // It went without a word, which only a panic in it does; the + // peer is as good as one that stopped reading. + None => SessionError::SendHoldTimerExpired(*self.send_hold.borrow()), + } + } + + /// Hands the writer whatever is still to go out, a NOTIFICATION among + /// it, and waits a moment for it to be written. The writer closes our + /// side of the connection once it has. + async fn close(mut self) { + self.flush(); + let Self { + writer, + writer_task, + mut pending, + .. + } = self; + let hand_over = async { + for batch in pending.drain(..) { + if writer.send(batch).await.is_err() { + break; + } + } + }; + let _ = timeout(CLOSE_LINGER, hand_over).await; + drop(writer); + let _ = timeout(CLOSE_LINGER, writer_task).await; + } +} + /// Writes a session's batches to the peer, as its own task, so that the /// session's reading never waits on its writing. Two routers sending each /// other their tables at once would otherwise each stop reading while @@ -1565,7 +1709,7 @@ async fn from_server(established: &mut Option, may_take_outbound: b /// through `failed`; the remaining batches are dropped with it. Once the /// session has nothing more to send, our side of the connection is shut. async fn write_to_peer( - mut writer: OwnedWriteHalf, + mut writer: impl AsyncWrite + Unpin, mut queue: mpsc::Receiver, send_hold: watch::Receiver, failed: mpsc::Sender, @@ -1784,17 +1928,9 @@ mod tests { )); } - /// A bug in the session must not leave the peer talking to a connection - /// nobody reads: it is told the session is being reset. - #[tokio::test] - async fn a_session_that_panicked_still_says_goodbye() { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let mut peer = TcpStream::connect(listener.local_addr().unwrap()) - .await - .unwrap(); - let (stream, peer_addr) = listener.accept().await.unwrap(); - let rib = Arc::new(RibTable::new(4)); - let server = Arc::new(ServerContext { + /// What every session of a test server shares, with `rib` as its table. + fn test_context(rib: Arc) -> Arc { + Arc::new(ServerContext { router_id: Ipv4Addr::new(10, 0, 0, 1), local_asn: 65000, rib: Arc::clone(&rib), @@ -1805,15 +1941,49 @@ mod tests { rejections: RejectionLog::default(), rpki: None, restarts: Restarts::new(rib, true, [], 120), - }); - let (mut session, _) = Session::open( + }) + } + + /// A session that an unconfigured peer has just connected to over the + /// loopback, and the peer's end of the connection. + async fn session_on_loopback() -> (Session, TcpStream) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let peer = TcpStream::connect(listener.local_addr().unwrap()) + .await + .unwrap(); + let (stream, peer_addr) = listener.accept().await.unwrap(); + let (session, _) = Session::open( stream, peer_addr, Direction::Inbound, - server, + test_context(Arc::new(RibTable::new(4))), PeerSettings::unconfigured(90, Arc::default(), None, GracefulRestartConfig::default()), ) .unwrap(); + (session, peer) + } + + /// The bytes of `msg` as they go to a peer. + fn wire(msg: &Message) -> BytesMut { + let mut out = BytesMut::new(); + msg.to_wire(&ParserContext::default(), &mut out).unwrap(); + out + } + + /// Takes every slot in the writer's queue, as batches the writer has + /// not got round to would; the slots are given back as the permits are + /// dropped. + fn fill_queue(outgoing: &Outgoing) -> Vec> { + (0..WRITER_QUEUE) + .map(|_| outgoing.writer.clone().try_reserve_owned().unwrap()) + .collect() + } + + /// A bug in the session must not leave the peer talking to a connection + /// nobody reads: it is told the session is being reset. + #[tokio::test] + async fn a_session_that_panicked_still_says_goodbye() { + let (mut session, mut peer) = session_on_loopback().await; let error = session .close_after_panic("index out of bounds".into()) @@ -1838,34 +2008,12 @@ mod tests { /// nothing, timers included, while it waits to write. #[tokio::test] async fn a_peer_that_stops_reading_is_given_up_on() { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - // Held, and never read from. - let _peer = TcpStream::connect(listener.local_addr().unwrap()) - .await - .unwrap(); - let (stream, peer_addr) = listener.accept().await.unwrap(); - let rib = Arc::new(RibTable::new(4)); - let server = Arc::new(ServerContext { - router_id: Ipv4Addr::new(10, 0, 0, 1), - local_asn: 65000, - rib: Arc::clone(&rib), - registry: SessionRegistry::new(), - collisions: CollisionTable::new(false), - hold_downs: HoldDowns::default(), - topology: None, - rejections: RejectionLog::default(), - rpki: None, - restarts: Restarts::new(rib, true, [], 120), - }); - let (mut session, _) = Session::open( - stream, - peer_addr, - Direction::Inbound, - server, - PeerSettings::unconfigured(90, Arc::default(), None, GracefulRestartConfig::default()), - ) - .unwrap(); - session.send_hold.send_replace(Duration::from_millis(100)); + // The peer's end is held, and never read from. + let (mut session, _peer) = session_on_loopback().await; + session + .outgoing + .send_hold + .send_replace(Duration::from_millis(100)); // Until the socket's buffers are full, writes go through; then // the writer gives up, and says so. @@ -1876,8 +2024,8 @@ mod tests { let stuck = tokio::time::timeout(Duration::from_secs(30), async { loop { session.send(&filler).unwrap(); - session.flush(); - if let Ok(e) = session.write_failed.try_recv() { + session.outgoing.flush(); + if let Ok(e) = session.outgoing.write_failed.try_recv() { return e; } tokio::time::sleep(Duration::from_millis(1)).await; @@ -1892,6 +2040,114 @@ mod tests { assert!(stuck.notification().is_none(), "there is no way to say so"); } + /// A session never waits to write, since it would not be reading + /// meanwhile: what the writer has no room for waits here instead, and + /// while it does, nothing more is taken from the rest of the server. + #[tokio::test] + async fn batches_wait_while_the_writer_has_no_room() { + let (_peer, ours) = duplex(4096); + let mut outgoing = Outgoing::new(ours); + let taken = fill_queue(&outgoing); + assert!(outgoing.may_take_more()); + + outgoing.send(&Message::KeepAlive).unwrap(); + assert!(outgoing.pending.is_empty(), "a batch is made on flush"); + outgoing.flush(); + assert_eq!(outgoing.pending.len(), 1); + assert!(!outgoing.may_take_more()); + + outgoing.send(&Message::KeepAlive).unwrap(); + outgoing.flush(); + assert_eq!(outgoing.pending.len(), 2); + drop(taken); + } + + /// Room in the writer's queue goes to the batch that has waited + /// longest, so that messages reach the peer in the order they were + /// sent. + #[tokio::test] + async fn room_in_the_writer_goes_to_the_oldest_batch() { + let (mut peer, ours) = duplex(4096); + let mut outgoing = Outgoing::new(ours); + let mut taken = fill_queue(&outgoing); + let first = Message::KeepAlive; + let second = Message::Notification(NotificationMessage { + error: BgpError::Cease(CeaseSubcode::AdminShutdown), + data: vec![1, 2, 3], + }); + outgoing.send(&first).unwrap(); + outgoing.flush(); + outgoing.send(&second).unwrap(); + outgoing.flush(); + assert_eq!(outgoing.pending.len(), 2); + + // One slot comes free. + drop(taken.pop()); + let permit = outgoing + .writer_says() + .await + .expect("the writer has room and has not failed"); + outgoing.use_room(permit); + assert_eq!(outgoing.pending.len(), 1); + assert_eq!(outgoing.pending.front(), Some(&wire(&second))); + assert!(!outgoing.may_take_more()); + + drop(taken); + outgoing.close().await; + let mut received = Vec::new(); + peer.read_to_end(&mut received).await.unwrap(); + assert_eq!(received, [wire(&first), wire(&second)].concat()); + } + + /// A piece from the exporter is counted by the messages in it, which + /// are found by their headers, since the exporter does not say. + #[tokio::test] + async fn a_piece_is_counted_as_the_messages_in_it() { + let (_peer, ours) = duplex(4096); + let mut outgoing = Outgoing::new(ours); + let updates = MESSAGES_SENT.with_label_values(&["update"]); + let before = updates.get(); + + let mut piece = BytesMut::new(); + for _ in 0..3 { + piece.extend_from_slice(&KEEPALIVE); + } + outgoing.send_piece(piece); + assert_eq!(updates.get() - before, 3); + assert_eq!(outgoing.write_buf.len(), 3 * KEEPALIVE.len()); + } + + /// When the connection goes while batches are waiting for the writer, + /// the session ends for the connection's loss and not for a send hold + /// timer: the peer's routes are kept over the one and not the other. + #[tokio::test] + async fn a_lost_connection_is_reported_as_one_while_batches_wait() { + // Too small for one message, and never read from: the writer + // stalls on the first batch. + let (peer, ours) = duplex(16); + let mut outgoing = Outgoing::new(ours); + outgoing.send(&Message::KeepAlive).unwrap(); + outgoing.flush(); + while outgoing.writer.capacity() < WRITER_QUEUE { + tokio::task::yield_now().await; + } + for _ in 0..=WRITER_QUEUE { + outgoing.send(&Message::KeepAlive).unwrap(); + outgoing.flush(); + } + assert!(!outgoing.may_take_more()); + + drop(peer); + let error = outgoing + .writer_says() + .await + .expect_err("there is no room in a writer that is gone"); + assert!( + matches!(&error, SessionError::Io(e) if e.kind() == io::ErrorKind::BrokenPipe), + "{error}" + ); + } + /// A session task that is torn down without running its cleanup must /// not leave the peer's routes behind. #[tokio::test] diff --git a/crates/server/src/dialer.rs b/crates/server/src/dialer.rs index a73211b..26fc76c 100644 --- a/crates/server/src/dialer.rs +++ b/crates/server/src/dialer.rs @@ -18,7 +18,7 @@ use tracing::{info, warn}; use crate::collision::Direction; use crate::config::PeerConfig; -use crate::connection::{CLOSED_AT_OUR_REQUEST, PeerSettings, ServerContext, run_session}; +use crate::connection::{PeerSettings, ServerContext, run_session}; use crate::session::DialState; /// Upper bound for exponential backoff between connection attempts. @@ -102,7 +102,9 @@ async fn dial_loop(peer: PeerConfig, server: Arc, connect_retry: registry.set_dial_state(&peer.name, DialState::Connecting); info!(peer = %peer.name, endpoint = %peer.endpoint, "Connecting to peer"); - let ended = match tokio::time::timeout(CONNECT_TIMEOUT, connect(&peer)).await { + // Why the attempt ended, and whether that is a failure worth a + // line in the log. + let (reason, failed) = match tokio::time::timeout(CONNECT_TIMEOUT, connect(&peer)).await { Ok(Ok(stream)) => match stream.peer_addr() { Ok(peer_addr) => { info!(peer = %peer.name, endpoint = %peer.endpoint, "TCP connected, starting session"); @@ -121,25 +123,17 @@ async fn dial_loop(peer: PeerConfig, server: Arc, connect_retry: if end.established { backoff = connect_retry; } - if end.result.is_ok() { - Ok(()) - } else { - Err(end.reason()) - } + let failed = end.result.is_err(); + (end.reason(), failed) } - Err(e) => Err(format!("connection lost at once: {e}")), + Err(e) => (format!("connection lost at once: {e}"), true), }, - Ok(Err(e)) => Err(format!("connect failed: {e}")), - Err(_) => Err(format!("connect timed out after {CONNECT_TIMEOUT:?}")), - }; - - let reason = match ended { - Ok(()) => CLOSED_AT_OUR_REQUEST.to_string(), - Err(reason) => { - warn!(peer = %peer.name, endpoint = %peer.endpoint, err = %reason, "Session attempt ended"); - reason - } + Ok(Err(e)) => (format!("connect failed: {e}"), true), + Err(_) => (format!("connect timed out after {CONNECT_TIMEOUT:?}"), true), }; + if failed { + warn!(peer = %peer.name, endpoint = %peer.endpoint, err = %reason, "Session attempt ended"); + } // Not all at the same moment, when many peers went away together. let wait = jittered(backoff); diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index daf5177..a2496d6 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -41,6 +41,9 @@ pub enum SessionError { Decode(#[from] DecodeError), #[error("could not encode a message: {0}")] Encode(#[from] EncodeError), + /// A received UPDATE's attributes could not be kept in the table. + #[error(transparent)] + Unstorable(#[from] crate::routes::Unstorable), /// The peer sent nothing for as long as the pre-negotiation hold time. #[error("timed out waiting for a message in the {0} state")] Timeout(State), @@ -101,9 +104,10 @@ impl SessionError { pub fn notification(&self) -> Option { let error = match self { Self::Io(_) | Self::SendHoldTimerExpired(_) | Self::NotifiedByPeer(_) => return None, - // Something of ours could not be written out, which is a bug - // here: as for a panic, the session is reset. - Self::Encode(_) => BgpError::Cease(CeaseSubcode::AdminReset), + // Something of ours could not be written out, or the peer's + // routes could not be kept, which is a bug here: as for a + // panic, the session is reset. + Self::Encode(_) | Self::Unstorable(_) => BgpError::Cease(CeaseSubcode::AdminReset), Self::Decode(e) => { return Some(NotificationMessage { error: e.notification.clone(), diff --git a/crates/server/src/restart.rs b/crates/server/src/restart.rs index 5554e62..46da267 100644 --- a/crates/server/src/restart.rs +++ b/crates/server/src/restart.rs @@ -244,6 +244,30 @@ pub struct Kept { pub until: SystemTime, } +impl Kept { + /// What was kept of a peer: `routes` routes of the IP versions + /// `versions`, under `rib_peer`, of the peer at `addr` called `name`. + /// Which restart they are of and until when they are kept are the + /// table's to say, in [`Restarts::keep`]. + pub(crate) fn new( + addr: SocketAddr, + name: Option, + rib_peer: PeerIndex, + versions: Vec, + routes: usize, + ) -> Self { + Self { + id: 0, + addr, + rib_peer, + name, + versions, + routes, + until: SystemTime::UNIX_EPOCH, + } + } +} + /// What there is of graceful restart while the server runs: the routes /// kept of peers that are restarting, and how far the peers are with /// sending their routes after this daemon started. @@ -298,33 +322,14 @@ impl Restarts { }) } - /// Notes that the routes of `peer` under `rib_peer` have been kept - /// (`routes` of them, of the IP versions `versions`) until `for_at_most` - /// from now, and has them removed then unless the peer is back. - #[allow(clippy::too_many_arguments)] - pub(crate) fn keep( - self: &Arc, - peer: PeerKey, - addr: SocketAddr, - name: Option, - rib_peer: PeerIndex, - versions: Vec, - routes: usize, - for_at_most: Duration, - ) { + /// Notes that `kept` says what routes of `peer` have been kept, until + /// `for_at_most` from now, and has them removed then unless the peer + /// is back. + pub(crate) fn keep(self: &Arc, peer: PeerKey, mut kept: Kept, for_at_most: Duration) { let id = self.ids.fetch_add(1, Ordering::Relaxed); - self.kept.locked().insert( - peer, - Kept { - id, - addr, - rib_peer, - name, - versions, - routes, - until: SystemTime::now() + for_at_most, - }, - ); + kept.id = id; + kept.until = SystemTime::now() + for_at_most; + self.kept.locked().insert(peer, kept); let restarts = Arc::clone(self); tokio::spawn(async move { tokio::time::sleep(for_at_most).await; diff --git a/crates/server/src/routes.rs b/crates/server/src/routes.rs index 215a2d8..9f74594 100644 --- a/crates/server/src/routes.rs +++ b/crates/server/src/routes.rs @@ -21,6 +21,14 @@ use crate::rpki::Checks; /// than a peer sending the whole Internet has; a bound all the same. const MAX_HELD: usize = 200_000; +/// Why an UPDATE could not be applied: the attributes its routes were to +/// be kept with could not be written in the form the table keeps them in. +/// Attributes that were decoded from a message always can be, so this is +/// a bug here, and not anything the peer did. +#[derive(Debug, thiserror::Error)] +#[error("could not keep the routes' attributes: {0}")] +pub struct Unstorable(#[from] pub(crate) EncodeError); + /// What applying an UPDATE did. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub(crate) struct Applied { @@ -113,7 +121,7 @@ pub(crate) fn apply( held: &mut Held, update: &UpdateMessage, attributes_wire: Option<&[u8]>, -) -> Result { +) -> Result { let mut applying = Applying { table, import, @@ -598,3 +606,254 @@ pub(crate) fn revalidate( } applied } + +#[cfg(test)] +mod tests { + use std::net::{IpAddr, Ipv4Addr}; + + use bgp_packet::message::{AsPathAttribute, NextHopPathAttribute, OriginPathAttribute}; + use bgp_rib::table::PeerInfo; + use bgp_rpki::{Vrp, VrpTable}; + + use super::*; + use crate::rejections::RejectionLog; + + const PEER_ASN: u32 = 65001; + const PEER_ADDRESS: Ipv4Addr = Ipv4Addr::new(192, 0, 2, 1); + + fn prefix(text: &str) -> IpPrefix { + IpPrefix::try_from(text).unwrap() + } + + fn nlri(text: &str) -> Nlri { + prefix(text).into() + } + + fn attrs_of(route: &str) -> Arc { + RibTable::new(1) + .attrs() + .intern(&announcement(route).path_attributes, None) + .unwrap() + } + + /// An UPDATE as the peer announces `route`, originating it itself. + fn announcement(route: &str) -> UpdateMessage { + UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![ + PathAttribute::Origin(OriginPathAttribute::IGP), + AsPathAttribute::from_asns(vec![PEER_ASN]), + PathAttribute::NextHop(NextHopPathAttribute(PEER_ADDRESS)), + ], + announced_routes: vec![nlri(route)], + } + } + + /// VRPs that allow `route` to be originated by `asn` and nobody else. + fn vrps(route: &str, asn: u32) -> VrpTable { + let prefix = prefix(route); + VrpTable::new([Vrp { + prefix, + max_length: prefix.length(), + asn, + }]) + } + + /// A table with the peer in it, and what is needed to apply its + /// UPDATEs by a permissive policy, checking only against `vrps`. + struct Fixture { + table: RibTable, + peer: PeerIndex, + policy: ImportPolicy, + rejections: RejectionLog, + } + + impl Fixture { + fn new() -> Self { + let table = RibTable::new(1); + let peer = table + .add_peer(PeerInfo { + asn: PEER_ASN, + router_id: Ipv4Addr::new(10, 0, 0, 2), + address: PEER_ADDRESS.into(), + interface: 0, + ibgp: false, + reflector_client: false, + }) + .unwrap(); + Self { + table, + peer, + policy: ImportPolicy::permissive(), + rejections: RejectionLog::default(), + } + } + + fn with_import( + &self, + vrps: Option<&VrpTable>, + held: &mut Held, + work: impl FnOnce(&RibTable, &Import<'_>, &mut Held) -> T, + ) -> T { + let import = Import { + peer: self.peer, + from: RouteSource { + peer_asn: PEER_ASN, + local_asn: 65000, + ibgp: false, + four_octet_asn: true, + role: None, + peer_address: PEER_ADDRESS.into(), + local_address: Ipv4Addr::new(192, 0, 2, 9).into(), + extended_next_hop: false, + }, + policy: &self.policy, + checks: Checks { + vrps, + ..Checks::NONE + }, + rejections: &self + .rejections + .of_peer(IpAddr::from(PEER_ADDRESS).into(), "test"), + }; + work(&self.table, &import, held) + } + + fn announce(&self, vrps: Option<&VrpTable>, held: &mut Held, route: &str) -> Applied { + self.with_import(vrps, held, |table, import, held| { + apply(table, import, held, &announcement(route), None).unwrap() + }) + } + + fn revalidate( + &self, + vrps: Option<&VrpTable>, + held: &mut Held, + suspects: &[Nlri], + ) -> Applied { + self.with_import(vrps, held, |table, import, held| { + revalidate(table, import, held, suspects) + }) + } + } + + /// Past the bound, a new route is dropped and the overflow noted, so + /// that the peer can be asked for it later; one already held is still + /// replaced, since that takes no more room. + #[test] + fn holding_stops_at_the_bound_and_says_so() { + let mut held = Held::default(); + let attrs = attrs_of("198.51.100.0/24"); + for n in 0..MAX_HELD as u32 { + let [a, b, c, d] = (0x0a00_0000 + n).to_be_bytes(); + held.hold( + IpPrefix::new(IpVersion::Ipv4, [a, b, c, d], 32) + .unwrap() + .into(), + &attrs, + ); + } + assert_eq!(held.len(), MAX_HELD); + assert!(!held.take_overflowed()); + + held.hold(nlri("198.51.100.0/24"), &attrs); + assert_eq!(held.len(), MAX_HELD); + assert!(held.take_overflowed()); + assert!( + !held.take_overflowed(), + "noted once, until it happens again" + ); + + let other = attrs_of("203.0.113.0/24"); + held.hold(nlri("10.0.0.0/32"), &other); + assert_eq!(held.len(), MAX_HELD); + assert!(!held.take_overflowed()); + assert!(Arc::ptr_eq(&held.routes[&nlri("10.0.0.0/32")], &other)); + } + + #[test] + fn forgetting_takes_out_what_is_named_and_nothing_else() { + let mut held = Held::default(); + let attrs = attrs_of("198.51.100.0/24"); + held.hold(nlri("198.51.100.0/24"), &attrs); + held.hold(nlri("203.0.113.0/24"), &attrs); + + held.forget(&[nlri("198.51.100.0/24"), nlri("192.0.2.0/24")]); + assert_eq!(held.len(), 1); + assert!(held.routes.contains_key(&nlri("203.0.113.0/24"))); + + held.forget(&[nlri("203.0.113.0/24")]); + held.forget(&[nlri("203.0.113.0/24")]); + assert_eq!(held.len(), 0); + } + + /// A route the VRPs find invalid stays out of the table but is not + /// lost: when the VRPs change in its favour it goes in, without the + /// peer having to send it again. + #[test] + fn a_held_route_is_released_once_the_checks_pass_it() { + let fixture = Fixture::new(); + let mut held = Held::default(); + let route = "198.51.100.0/24"; + let against = vrps(route, PEER_ASN + 1); + let applied = fixture.announce(Some(&against), &mut held, route); + assert_eq!(applied.rejected, 1); + assert_eq!(applied.outcome.added, 0); + assert!(fixture.table.is_empty()); + assert_eq!(held.len(), 1); + + // Nothing has changed in the route's favour yet. + let applied = fixture.revalidate(Some(&against), &mut held, &[]); + assert_eq!(applied, Applied::default()); + assert_eq!(held.len(), 1); + + let for_it = vrps(route, PEER_ASN); + let applied = fixture.revalidate(Some(&for_it), &mut held, &[]); + assert_eq!(applied.outcome.added, 1); + assert_eq!(applied.rejected, 0); + assert_eq!(held.len(), 0); + assert_eq!(fixture.table.routes(&prefix(route)).len(), 1); + } + + /// The other way round: a route in the table that the VRPs come to + /// find invalid is taken out and held, and comes back once they + /// change again. + #[test] + fn a_route_turned_invalid_is_held_and_comes_back() { + let fixture = Fixture::new(); + let mut held = Held::default(); + let route = "198.51.100.0/24"; + let for_it = vrps(route, PEER_ASN); + let applied = fixture.announce(Some(&for_it), &mut held, route); + assert_eq!(applied.outcome.added, 1); + assert_eq!(fixture.table.len(), 1); + + let against = vrps(route, PEER_ASN + 1); + let applied = fixture.revalidate(Some(&against), &mut held, &[nlri(route)]); + assert_eq!(applied.rejected, 1); + assert_eq!(applied.outcome.removed, 1); + assert!(fixture.table.is_empty()); + assert_eq!(held.len(), 1); + + let applied = fixture.revalidate(Some(&for_it), &mut held, &[]); + assert_eq!(applied.outcome.added, 1); + assert_eq!(held.len(), 0); + assert_eq!(fixture.table.len(), 1); + } + + /// Without anything to check against there is nothing to look at + /// again: what is held stays held, and what is in the table stays. + #[test] + fn revalidation_without_checks_changes_nothing() { + let fixture = Fixture::new(); + let mut held = Held::default(); + let route = "198.51.100.0/24"; + fixture.announce(Some(&vrps(route, PEER_ASN + 1)), &mut held, route); + assert_eq!(held.len(), 1); + + let applied = fixture.revalidate(None, &mut held, &[nlri(route)]); + assert_eq!(applied, Applied::default()); + assert_eq!(held.len(), 1); + assert!(fixture.table.is_empty()); + } +} diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index e0f5518..2618c7c 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -237,7 +237,7 @@ impl BgpServer { }; match result { Ok((stream, peer_addr)) => { - match admit(&self.config, &context, peer_addr.into(), &connections) { + match screen(&self.config, &context, peer_addr.into(), &connections) { Ok((peer, slot)) => { let context = Arc::clone(&context); tokio::spawn(serve(stream, peer_addr, context, peer, slot)); @@ -461,9 +461,10 @@ fn listen(addr: SocketAddr) -> io::Result { socket.listen(LISTEN_BACKLOG) } -/// Decides whether to take a connection from `address`, and if so with what -/// settings. -fn admit( +/// Screens a connection from `address`: whether to take it, and if so +/// with what settings. Not the collision check, which is +/// [`crate::collision::CollisionTable::admit`]. +fn screen( config: &Config, context: &ServerContext, address: PeerAddress,