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,