Write SSE frames eagerly and raise the send cap to 512 KiB

crates/beacon_api/src/server.rs

@@ -9,8 +9,8 @@ use mio::{Events, Interest, Registry, Token, event::Event};
 use silver_beacon_state_data::{BeaconStateReader, SpecConfig};
 use silver_common::{Enr, Identify, Keypair};
 use silver_httpcore::{
-    AfterResponse, Bind, ChunkedResponse, Listener, ParsedRequest, ServerConnection, Stream,
-    TokenRange,
+    AfterResponse, Bind, ChunkedResponse, Closed, Listener, ParsedRequest, ServerConnection,
+    Stream, TokenRange,
 };
 
 use crate::{
@@ -255,18 +255,7 @@ impl Subscription {
             return Ok(true);
         }
 
-        if event.is_writable() {
-            while !self.body.pending_write().is_empty() {
-                match stream.write(self.body.pending_write()) {
-                    Ok(0) => {
-                        return Err(io::Error::new(io::ErrorKind::WriteZero, "write returned 0"))
-                    }
-                    Ok(n) => self.body.commit_write(n, now),
-                    Err(e) if would_block(&e) => return Ok(false),
-                    Err(e) if interrupted(&e) => continue,
-                    Err(e) => return Err(e),
-                }
-            }
+        if event.is_writable() && self.body.drain_into(stream, now)? {
             registry.reregister(stream, event.token(), Interest::READABLE)?;
         }
 
@@ -435,26 +424,36 @@ impl BeaconApi {
         let Self { connections, registry, .. } = self;
         let mut pushed = false;
         connections.retain(|token, conn| {
-            let State::Subscription(subscription) = &mut conn.state else { return true };
+            let Connection { stream, state: State::Subscription(subscription) } = conn else {
+                return true;
+            };
             if !wants(subscription) {
                 return true;
             }
-            if !subscription.body.push(chunk, now) {
-                tracing::warn!(
-                    "beacon api subscriber would exceed send cap with {} bytes already pending, closing",
-                    subscription.body.pending_write().len()
-                );
-                let _ = registry.deregister(&mut conn.stream);
-                return false;
-            }
-            pushed = true;
-            let interest = Interest::READABLE | Interest::WRITABLE;
-            if let Err(e) = registry.reregister(&mut conn.stream, *token, interest) {
-                tracing::warn!("beacon api subscriber lost: {e}");
-                let _ = registry.deregister(&mut conn.stream);
-                return false;
+            let outcome = subscription.body.deliver(stream, chunk, now).and_then(|interest| {
+                pushed = true;
+                match interest {
+                    Some(interest) => {
+                        registry.reregister(stream, *token, interest).map_err(Closed::Lost)
+                    }
+                    None => Ok(()),
+                }
+            });
+            match outcome {
+                Ok(()) => true,
+                Err(Closed::AtCap { pending }) => {
+                    tracing::warn!(
+                        "beacon api subscriber would exceed send cap with {pending} bytes already pending, closing"
+                    );
+                    let _ = registry.deregister(stream);
+                    false
+                }
+                Err(Closed::Lost(e)) => {
+                    tracing::warn!("beacon api subscriber lost: {e}");
+                    let _ = registry.deregister(stream);
+                    false
+                }
             }
-            true
         });
         pushed
     }
@@ -1447,6 +1446,115 @@ mod tests {
         assert_same_bytes(&got, &expected);
     }
 
+    fn burst_frame(index: usize, len: usize) -> Vec<u8> {
+        let mut frame = format!("event: burst\ndata: {index:02}").into_bytes();
+        frame.resize(len - 2, b'c');
+        frame.extend_from_slice(b"\n\n");
+        frame
+    }
+
+    /// The burst fits in the application buffer even if the socket initially
+    /// accepts no bytes. Delivery therefore does not require a particular
+    /// kernel send-buffer capacity. The frames approximate one block's column
+    /// events with commitments; any unsent remainder drains through the
+    /// readiness loop.
+    #[test]
+    fn a_burst_reaches_a_reading_subscriber_in_order() {
+        let dir = tempfile::tempdir().unwrap();
+        let socket = dir.path().join("api.sock");
+        let mut server = server_bound_to(&[Bind::Unix(socket.clone())], 64, LONG_TIMEOUT);
+        let mut client = connect_uds(&socket);
+        subscribe(&mut client, "block");
+        pump_until(&mut server, "subscribed and head sent", |server| {
+            subscribers(server) == 1 && bytes_waiting_for_subscribers(server) == 0
+        });
+
+        let frames: Vec<_> = (0..128).map(|index| burst_frame(index, 2300)).collect();
+        let mut expected = SSE_HEAD.to_vec();
+        frames.iter().for_each(|frame| expected.extend(chunk(frame)));
+        let now = Instant::now();
+        for frame in &frames {
+            assert!(server.api.fan_out(|_| true, frame, now), "queued for the subscriber");
+        }
+        assert_eq!(subscribers(&server), 1);
+
+        let got = serve(&mut server, read_exactly(client, expected.len()), "the burst");
+        assert_same_bytes(&got, &expected);
+    }
+
+    /// Checks that publication attempts delivery before the readiness loop,
+    /// independently of whether a larger burst fits the application buffer.
+    #[test]
+    fn a_publish_to_a_reading_subscriber_leaves_nothing_pending() {
+        let dir = tempfile::tempdir().unwrap();
+        let socket = dir.path().join("api.sock");
+        let mut server = server_bound_to(&[Bind::Unix(socket.clone())], 64, LONG_TIMEOUT);
+        let mut client = connect_uds(&socket);
+        subscribe(&mut client, "block");
+        pump_until(&mut server, "subscribed and head sent", |server| {
+            subscribers(server) == 1 && bytes_waiting_for_subscribers(server) == 0
+        });
+
+        for slot in 1..=3 {
+            server.api.publish_block(slot, &[0x33; 32]);
+        }
+        assert_eq!(bytes_waiting_for_subscribers(&server), 0);
+        assert_eq!(subscribers(&server), 1);
+        drop(client);
+    }
+
+    /// On Linux, registering `WRITABLE` on a writable socket queues an epoll
+    /// event. The small frame keeps this probe independent of burst capacity.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_frame_written_whole_to_an_idle_subscriber_leaves_nothing_to_report() {
+        let mut server = server_with(64, LONG_TIMEOUT);
+        let mut client = connect(tcp_addr(&server));
+        subscribe(&mut client, "block");
+        pump_until(&mut server, "subscribed and head sent", |server| {
+            subscribers(server) == 1 && bytes_waiting_for_subscribers(server) == 0
+        });
+
+        server.api.publish_block(7, &[0x77; 32]);
+        assert_eq!(bytes_waiting_for_subscribers(&server), 0, "the socket took the frame");
+        server.readiness.wait(Duration::ZERO);
+        assert_eq!(server.readiness.events().iter().count(), 0);
+        drop(client);
+    }
+
+    /// The queued response head leaves `WRITABLE` registered. On Linux, a
+    /// publish that drains the head must remove that interest before polling
+    /// and retain interest in inbound bytes.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_publish_that_drains_the_unsent_head_leaves_nothing_to_report() {
+        let dir = tempfile::tempdir().unwrap();
+        let socket = dir.path().join("api.sock");
+        let mut server = server_bound_to(&[Bind::Unix(socket.clone())], 64, LONG_TIMEOUT);
+        let mut client = connect_uds(&socket);
+        subscribe(&mut client, "block");
+        pump_until(&mut server, "subscribed", |server| subscribers(server) == 1);
+        assert!(bytes_waiting_for_subscribers(&server) > 0, "the head is still queued");
+
+        server.api.publish_block(5, &[0x55; 32]);
+        assert_eq!(bytes_waiting_for_subscribers(&server), 0, "the publish drained the head");
+        server.readiness.wait(Duration::ZERO);
+        assert_eq!(server.readiness.events().iter().count(), 0, "no writable event remains");
+
+        client.write_all(b"GET /eth/v1/node/version HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();
+        assert!(server.pump(), "inbound bytes produce a readable event");
+        assert_eq!(
+            subscribers(&server),
+            1,
+            "a request behind the subscribe is dropped, not answered"
+        );
+
+        drop(client);
+        pump_until(&mut server, "hung-up subscriber removed", |server| {
+            server.api.connections.is_empty()
+        });
+    }
+
     #[test]
     fn a_subscriber_that_stops_reading_is_closed_at_the_cap() {
         let dir = tempfile::tempdir().unwrap();