Merge pull request #4382 from cbeck88/fix/gateway-outbound-queues

Fix/gateway outbound queues

Author: vicsn

node/bft/events/src/lib.rs

@@ -119,10 +119,96 @@ impl<N: Network> From<DisconnectReason> for Event<N> {
     }
 }
 
+/// Replaces the payload with its serialized form, if it is not already serialized.
+fn serialize_payload<T: FromBytes + ToBytes + Send + 'static>(payload: &mut Data<T>) -> Result<()> {
+    if let Data::Object(object) = payload {
+        *payload = Data::Buffer(object.to_bytes_le()?.into());
+    }
+    Ok(())
+}
+
+/// Returns `true` if the payload still holds an object that would have to be serialized.
+fn is_payload_unserialized<T: FromBytes + ToBytes + Send + 'static>(payload: &mut Data<T>) -> bool {
+    matches!(payload, Data::Object(_))
+}
+
+/// Applies `$f` to the event's [`Data`] payload, evaluating to `$default` if it carries none.
+///
+/// Both [`Event::serialize_payload`] and [`Event::has_unserialized_payload`] need to know which
+/// variants carry a payload. Routing both through this macro keeps that list in exactly one place,
+/// so a newly added variant cannot be handled by one and silently forgotten by the other.
+macro_rules! with_data_payload {
+    ($event:expr, $f:ident, $default:expr) => {
+        match $event {
+            Self::BatchPropose(event) => $f(&mut event.batch_header),
+            Self::BatchCertified(event) => $f(&mut event.certificate),
+            Self::PrimaryPing(event) => $f(&mut event.primary_certificate),
+            Self::BlockResponse(event) => $f(&mut event.blocks),
+            Self::ChallengeResponse(event) => $f(&mut event.signature),
+            Self::TransmissionResponse(event) => match &mut event.transmission {
+                Transmission::Solution(solution) => $f(solution),
+                Transmission::Transaction(transaction) => $f(transaction),
+                Transmission::Ratification => $default,
+            },
+            // The remaining events carry no `Data` payload.
+            //
+            // Note that `CertificateResponse` is in this list only because its certificate is held
+            // directly rather than behind a `Data`, unlike every other certificate-bearing event.
+            // It is consequently serialized on the writer task and deserialized on the reading
+            // task, and cannot benefit from either of the methods below until that is changed.
+            Self::BatchSignature(_)
+            | Self::BlockRequest(_)
+            | Self::CertificateRequest(_)
+            | Self::CertificateResponse(_)
+            | Self::ChallengeRequest(_)
+            | Self::Disconnect(_)
+            | Self::TransmissionRequest(_)
+            | Self::ValidatorsRequest(_)
+            | Self::ValidatorsResponse(_)
+            | Self::WorkerPing(_) => $default,
+        }
+    };
+}
+
 impl<N: Network> Event<N> {
     /// The version of the event protocol; it can be incremented in order to force users to update.
     pub const VERSION: u32 = 10;
 
+    /// Serializes the event's [`Data`] payload, if it holds one that is not already serialized.
+    ///
+    /// [`Data`] defers serialization until the event is written to a stream, so an event that is
+    /// cloned for several peers is serialized once per peer, separately, inside each connection's
+    /// writer task. Calling this beforehand performs the serialization once; every clone then
+    /// shares the resulting buffer, and each writer only has to copy it to the wire.
+    ///
+    /// Before a broadcast, this avoids serializing the same payload once per recipient. Before a
+    /// single send it saves no total work, but it still matters: it moves the serialization off
+    /// the connection's writer task, which is a Tokio worker and should not be running compute.
+    /// A large payload serialized there stalls the reactor, and it does so inside the write
+    /// timeout, which then has to be sized to accommodate it.
+    ///
+    /// Accordingly, `Transport::send` performs this on a blocking thread for every outbound event,
+    /// and `Transport::broadcast` performs it once up front so that the fan-out only clones the
+    /// resulting buffer.
+    ///
+    /// It is a no-op for events that carry no payload, or whose payload is already serialized, so
+    /// applying it twice is harmless.
+    pub fn serialize_payload(&mut self) -> Result<()> {
+        with_data_payload!(self, serialize_payload, Ok(()))
+    }
+
+    /// Returns `true` if this event carries a payload that has not been serialized yet.
+    ///
+    /// Callers use this to decide whether [`Self::serialize_payload`] is worth the cost of moving
+    /// the event to a blocking thread. Most events carry no payload at all, and re-sending an
+    /// already-serialized one is common, so the check keeps that hop off the common path.
+    ///
+    /// Note this takes `&mut self` only because it shares its variant list with
+    /// [`Self::serialize_payload`]; it does not modify the event.
+    pub fn has_unserialized_payload(&mut self) -> bool {
+        with_data_payload!(self, is_payload_unserialized, false)
+    }
+
     /// Returns the event name.
     #[inline]
     pub fn name(&self) -> Cow<'static, str> {
@@ -257,15 +343,20 @@ pub mod prop_tests {
         Disconnect,
         DisconnectReason,
         Event,
+        ValidatorsRequest,
         batch_certified::prop_tests::any_batch_certified,
         batch_propose::prop_tests::any_batch_propose,
         batch_signature::prop_tests::any_batch_signature,
+        block_request::prop_tests::any_block_request,
+        block_response::prop_tests::any_block_response,
         certificate_request::prop_tests::any_certificate_request,
         certificate_response::prop_tests::any_certificate_response,
         challenge_request::prop_tests::any_challenge_request,
         challenge_response::prop_tests::any_challenge_response,
+        primary_ping::prop_tests::any_primary_ping,
         transmission_request::prop_tests::any_transmission_request,
         transmission_response::prop_tests::any_transmission_response,
+        validators_response::prop_tests::any_validators_response,
         worker_ping::prop_tests::any_worker_ping,
     };
     use snarkvm::{
@@ -316,11 +407,18 @@ pub mod prop_tests {
         .boxed()
     }
 
+    /// A strategy covering every [`Event`] variant.
+    ///
+    /// Keep this exhaustive. Several properties are asserted over "any event", and a variant that
+    /// is missing here is silently untested rather than failing -- which is how `BlockResponse` and
+    /// `PrimaryPing`, the two largest `Data`-carrying events, went uncovered.
     pub fn any_event() -> BoxedStrategy<Event<CurrentNetwork>> {
         prop_oneof![
             any_batch_certified().prop_map(Event::BatchCertified),
             any_batch_propose().prop_map(Event::BatchPropose),
             any_batch_signature().prop_map(Event::BatchSignature),
+            any_block_request().prop_map(Event::BlockRequest),
+            any_block_response().prop_map(Event::BlockResponse),
             any_certificate_request().prop_map(Event::CertificateRequest),
             any_certificate_response().prop_map(Event::CertificateResponse),
             any_challenge_request().prop_map(Event::ChallengeRequest),
@@ -342,8 +440,11 @@ pub mod prop_tests {
                 any::<Selector>()
             )
                 .prop_map(|(reasons, selector)| Event::Disconnect(Disconnect::from(selector.select(reasons)))),
+            any_primary_ping().prop_map(Event::PrimaryPing),
             any_transmission_request().prop_map(Event::TransmissionRequest),
             any_transmission_response().prop_map(Event::TransmissionResponse),
+            Just(ValidatorsRequest).prop_map(Event::ValidatorsRequest),
+            any_validators_response().prop_map(Event::ValidatorsResponse),
             any_worker_ping().prop_map(Event::WorkerPing)
         ]
         .boxed()
@@ -358,4 +459,55 @@ pub mod prop_tests {
         assert_eq!(original.id(), deserialized.id());
         assert_eq!(original.name(), deserialized.name());
     }
+
+    /// Serializing the payload ahead of time must be invisible on the wire, otherwise doing it
+    /// before a broadcast would change what peers receive.
+    #[proptest]
+    fn serialize_payload_preserves_the_encoding(#[strategy(any_event())] original: Event<CurrentNetwork>) {
+        let mut expected = Vec::new();
+        Event::write_le(&original, &mut expected).unwrap();
+
+        let mut event = original.clone();
+        event.serialize_payload().unwrap();
+
+        let mut actual = Vec::new();
+        Event::write_le(&event, &mut actual).unwrap();
+
+        assert_eq!(expected, actual, "{} encoded differently once its payload was serialized", original.name());
+    }
+
+    /// Serializing the payload must be idempotent, so that broadcasting an event that has already
+    /// been serialized does not deserialize and re-serialize it.
+    #[proptest]
+    fn serialize_payload_is_idempotent(#[strategy(any_event())] original: Event<CurrentNetwork>) {
+        let mut once = original.clone();
+        once.serialize_payload().unwrap();
+
+        let mut twice = once.clone();
+        twice.serialize_payload().unwrap();
+
+        assert_eq!(once, twice);
+    }
+
+    /// `has_unserialized_payload` is what decides whether an event is worth handing to a blocking
+    /// thread, so it must agree with `serialize_payload` about which events have work to do. If the
+    /// two ever disagreed, a payload-carrying event could be serialized on a writer task after all.
+    #[proptest]
+    fn has_unserialized_payload_agrees_with_serialize_payload(
+        #[strategy(any_event())] original: Event<CurrentNetwork>,
+    ) {
+        let mut event = original.clone();
+
+        // Serializing must clear the flag, whether or not it was set to begin with.
+        event.serialize_payload().unwrap();
+        assert!(!event.has_unserialized_payload(), "{} still reports an unserialized payload", original.name());
+
+        // And an event that reports no work to do must be unchanged by doing the work.
+        let mut reported_no_work = original.clone();
+        if !reported_no_work.has_unserialized_payload() {
+            let before = reported_no_work.clone();
+            reported_no_work.serialize_payload().unwrap();
+            assert_eq!(before, reported_no_work, "{} changed despite reporting no work", original.name());
+        }
+    }
 }

node/bft/src/gateway.rs

@@ -18,6 +18,7 @@ use crate::helpers::Telemetry;
 use crate::{
     CONTEXT,
     MAX_BATCH_DELAY,
+    MAX_FETCH_TIMEOUT,
     MEMORY_POOL_PORT,
     Worker,
     events::{DisconnectReason, EventCodec, PrimaryPing},
@@ -120,6 +121,35 @@ const CACHE_EVENTS_INTERVAL: i64 = (MAX_BATCH_DELAY.as_secs()) as i64; // second
 /// The maximum interval of requests to cache.
 const CACHE_REQUESTS_INTERVAL: i64 = (MAX_BATCH_DELAY.as_secs()) as i64; // seconds
 
+/// The number of consensus rounds of traffic that a per-connection message queue must absorb.
+///
+/// A message that has not made it onto (or off of) the wire within `MAX_FETCH_TIMEOUT` is already
+/// useless to the peer: a requester has given up on it by then (see
+/// `Worker::send_transmission_request`), and any consensus event it carried is stale. Sizing the
+/// per-connection queues to this window therefore bounds them to the traffic that can still be
+/// delivered in time, rather than to the entire garbage-collection window (`MAX_GC_ROUNDS`, which
+/// is ~100 rounds, i.e. several minutes of traffic).
+const QUEUE_WINDOW_ROUNDS: usize = (MAX_FETCH_TIMEOUT.as_millis() / MAX_BATCH_DELAY.as_millis()) as usize;
+
+/// Computes the depth of the per-connection inbound and outbound message queues.
+///
+/// These queues are transient send/receive buffers, not backlogs: they only need to hold the
+/// traffic a peer can legitimately exchange with us over `QUEUE_WINDOW_ROUNDS` (see above).
+/// Per round, the worst case is every certificate in the round plus every transmission each of
+/// those certificates contains — that is, a peer that is missing an entire round and fetches all
+/// of it from us. The leading factor of 2 is headroom for the remaining, far smaller, event
+/// traffic (batch proposals and signatures, primary and worker pings, block and validator
+/// requests) and for requests that straddle a round boundary.
+///
+/// Note that each slot can hold a full `Transmission`, so this value is a direct multiplier on the
+/// heap a single peer can pin. It must stay small enough that `depth * max transmission size` is
+/// survivable on a commodity validator.
+fn per_connection_queue_depth<N: Network>() -> usize {
+    2 * QUEUE_WINDOW_ROUNDS
+        * N::LATEST_MAX_CERTIFICATES() as usize
+        * (BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH + 1)
+}
+
 /// The maximum number of connection attempts in an interval.
 #[cfg(not(test))]
 const MAX_CONNECTION_ATTEMPTS: usize = 10;
@@ -1316,7 +1346,32 @@ impl<N: Network> Transport<N> for Gateway<N> {
     /// This function returns as soon as the event is queued to be sent,
     /// without waiting for the actual delivery; instead, the caller is provided with a [`oneshot::Receiver`]
     /// which can be used to determine when and whether the event has been delivered.
-    async fn send(&self, peer_ip: SocketAddr, event: Event<N>) -> Option<oneshot::Receiver<io::Result<()>>> {
+    async fn send(&self, peer_ip: SocketAddr, mut event: Event<N>) -> Option<oneshot::Receiver<io::Result<()>>> {
+        // Serialize the payload here, rather than leaving it for the connection's writer task.
+        //
+        // `Data` defers serialization until the event is written to the stream, which puts it on
+        // the writer task -- a Tokio worker -- and inside the write timeout. For the large
+        // responses (`BlockResponse` in particular, which can carry `DataBlocks::MAXIMUM_NUMBER_
+        // OF_BLOCKS`) that is a substantial amount of compute on the reactor.
+        //
+        // The check keeps the hop off the common path: most events carry no payload, and a
+        // broadcast has already serialized its payload before fanning out here, so this is a no-op
+        // for every recipient after the first.
+        if event.has_unserialized_payload() {
+            let name = event.name();
+            event = match spawn_blocking!({
+                let mut event = event;
+                event.serialize_payload()?;
+                Ok(event)
+            }) {
+                Ok(event) => event,
+                Err(err) => {
+                    error!("{CONTEXT} Unable to serialize '{name}' for '{peer_ip}' - {err}");
+                    return None;
+                }
+            };
+        }
+
         macro_rules! send {
             ($self:ident, $cache_map:ident, $interval:expr, $freq:ident) => {{
                 // Rate limit the number of certificate requests sent to the peer.
@@ -1357,14 +1412,29 @@ impl<N: Network> Transport<N> for Gateway<N> {
     }
 
     /// Broadcasts the given event to all connected peers.
-    // TODO(ljedrz): the event should be checked for the presence of Data::Object, and
-    // serialized in advance if it's there.
-    fn broadcast(&self, event: Event<N>) {
+    fn broadcast(&self, mut event: Event<N>) {
         // Ensure there are connected peers.
         if self.number_of_connected_peers() > 0 {
             let self_ = self.clone();
             let connected_peers = self.connected_peers();
             tokio::spawn(async move {
+                // Serialize the event's payload once, rather than once per recipient; every
+                // recipient then shares the resulting buffer. `Transport::send` would otherwise do
+                // this separately for each peer below.
+                if event.has_unserialized_payload() {
+                    let name = event.name();
+                    event = match spawn_blocking!({
+                        let mut event = event;
+                        event.serialize_payload()?;
+                        Ok(event)
+                    }) {
+                        Ok(event) => event,
+                        Err(err) => {
+                            error!("{CONTEXT} Unable to serialize '{name}' for broadcast - {err}");
+                            return;
+                        }
+                    };
+                }
                 // Iterate through all connected peers.
                 for peer_ip in connected_peers {
                     // Send the event to the peer.
@@ -1408,12 +1478,11 @@ impl<N: Network> Reading for Gateway<N> {
         Ok(())
     }
 
-    /// Computes the depth of per-connection queues used to process inbound messages, sufficient to process the maximum expected load at any givent moment.
+    /// Computes the depth of per-connection queues used to process inbound messages, sufficient to process the maximum expected load at any given moment.
     /// The greater it is, the more inbound messages the node can enqueue, but a too large value can make the node more susceptible to DoS attacks.
+    /// See [`per_connection_queue_depth`] for the derivation.
     fn message_queue_depth(&self) -> usize {
-        2 * BatchHeader::<N>::MAX_GC_ROUNDS
-            * N::LATEST_MAX_CERTIFICATES() as usize
-            * BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH
+        per_connection_queue_depth::<N>()
     }
 }
 
@@ -1428,13 +1497,12 @@ impl<N: Network> Writing for Gateway<N> {
         Default::default()
     }
 
-    /// Computes the depth of per-connection queues used to send outbound messages, sufficient to process the maximum expected load at any givent moment.
-    /// The greater it is, the more outbound messages the node can enqueue. A too large value large value might obscure potential issues with your implementation
-    /// (like slow serialization) or network.
+    /// Computes the depth of per-connection queues used to send outbound messages, sufficient to process the maximum expected load at any given moment.
+    /// The greater it is, the more outbound messages the node can enqueue. A too large value might obscure potential issues with your implementation
+    /// (like slow serialization) or network, and lets a peer that stops reading its socket pin an unreasonable amount of our heap.
+    /// See [`per_connection_queue_depth`] for the derivation.
     fn message_queue_depth(&self) -> usize {
-        2 * BatchHeader::<N>::MAX_GC_ROUNDS
-            * N::LATEST_MAX_CERTIFICATES() as usize
-            * BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH
+        per_connection_queue_depth::<N>()
     }
 }
 
@@ -2333,6 +2401,31 @@ mod prop_tests {
 
     type CurrentNetwork = MainnetV0;
 
+    /// The per-connection queues are sized from the fetch-timeout window, not the GC window.
+    ///
+    /// Each slot can hold a full `Transmission`, so the depth is a direct multiplier on the heap a
+    /// single peer can pin by not reading its socket. Pin the derivation so a regression is loud.
+    #[test]
+    fn test_per_connection_queue_depth() {
+        use crate::gateway::{QUEUE_WINDOW_ROUNDS, per_connection_queue_depth};
+        use snarkvm::console::network::Network;
+
+        // The window is `MAX_FETCH_TIMEOUT` expressed in rounds.
+        assert_eq!(QUEUE_WINDOW_ROUNDS, 3);
+
+        let certificates = CurrentNetwork::LATEST_MAX_CERTIFICATES() as usize;
+        let transmissions = BatchHeader::<CurrentNetwork>::MAX_TRANSMISSIONS_PER_BATCH;
+        let depth = per_connection_queue_depth::<CurrentNetwork>();
+
+        assert_eq!(depth, 2 * QUEUE_WINDOW_ROUNDS * certificates * (transmissions + 1));
+
+        // It must cover a peer fetching every transmission of every certificate for the window...
+        assert!(depth >= QUEUE_WINDOW_ROUNDS * certificates * transmissions);
+        // ...but stay far below the old `MAX_GC_ROUNDS`-derived depth, which let one peer pin
+        // 400,000 transmissions.
+        assert!(depth < 2 * BatchHeader::<CurrentNetwork>::MAX_GC_ROUNDS * certificates * transmissions / 8);
+    }
+
     impl Debug for Gateway<CurrentNetwork> {
         fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
             // TODO implement Debug properly and move it over to production code