session: write to the peer from a task of its own

A session read from and wrote to its socket in one loop: while it was
writing, it was not reading. Writing waits when the peer's receive
window is closed, and the peer's window closes when the peer is not
reading, which it is not while it is writing to us and waiting on our
window in turn. Two routers sending each other every path at once, as
they do with ADD-PATH both ways after a restart, each stopped reading
within seconds and waited on the other until the send hold timer ended
both sessions eight minutes later. The status page's TCP row showed it:
both windows zero, the send queue full, the receive queue unread. The
batching of writes in the previous change made a session spend far
longer writing and so made this certain; it was possible before.

Now each session has a writer task. The session encodes its messages
into batches and hands them to the task through a bounded queue without
ever waiting: a batch the queue has no room for is kept, and while any
is kept the session takes no more messages from the rest of the server
(which then waits, as before, in its own task) but goes on reading the
peer. The writer task's failure, or a write that makes no progress for
the send hold time, comes back to the session as a local failure and
ends it. When the session ends, what is still to go out, a NOTIFICATION
among it, is handed over and the writer shuts our side once it has
written it.

Testing: a new integration test has one peer's table in the server and
a second peer send a table of its own while reading nothing of what the
server sends it, and checks that the server takes the second table all
the same; against the previous commit the test hangs until its timeout.
The send hold timer test now reads the writer's failure. The session
tests, the integration tests, the kernel tests and every lab phase
pass.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

crates/integration-tests/tests/deadlock.rs

@@ -0,0 +1,92 @@
+//! A peer that sends us a table while not reading what we send it must
+//! not stop us reading it: two routers sending each other their tables at
+//! once would otherwise each wait on the other for ever.
+
+use std::net::Ipv4Addr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use bgp_integration_tests::announcement::Announcement;
+use bgp_integration_tests::prefix;
+use bgp_integration_tests::test_peer::TestPeer;
+use bgp_integration_tests::test_server::TestServer;
+use bgp_server::policy::ExportPolicy;
+use eyre::Result;
+
+const LOCAL_ASN: u32 = 64601;
+const ROUTES: u32 = 100_000;
+
+fn nth(n: u32) -> bgp_primitives::ip_prefix::IpPrefix {
+    let [a, b, c, _] = (0x0a00_0000u32 + n * 256).to_be_bytes();
+    prefix(&format!("{a}.{b}.{c}.0/24"))
+}
+
+#[tokio::test]
+async fn a_peer_that_does_not_read_is_still_read() -> Result<()> {
+    let server = TestServer::start_with(LOCAL_ASN, |c| {
+        c.export = Arc::new(ExportPolicy {
+            learned_routes: true,
+            all_paths: false,
+            ..ExportPolicy::default()
+        });
+    })
+    .await?;
+    // One peer's table, for the server to pass on.
+    let mut first = server
+        .established(64701, Ipv4Addr::new(10, 0, 0, 1))
+        .await?;
+    // A route to an UPDATE, each with attributes of its own, as a real
+    // table has them: passing it on is then megabytes, not a few hundred
+    // UPDATEs that the socket buffers would take whole.
+    for i in 0..ROUTES {
+        first
+            .announce(
+                Announcement::of([nth(i).into()], &[64701, 64510])
+                    .with_communities(&[((i >> 16) as u16 + 1, i as u16)]),
+            )
+            .await?;
+    }
+    server.wait_for_routes(ROUTES as usize).await?;
+
+    // A second peer, which the server starts sending that table to as
+    // soon as it is up, sends a table of its own and reads nothing back:
+    // the server's writes to it fill its socket and stop. The server has
+    // to go on reading it all the same. The peer's receive buffer is
+    // small, as a real link's window would be, so that this happens with
+    // the first few hundred kilobytes and not after the loopback's
+    // megabytes.
+    let socket = tokio::net::TcpSocket::new_v4()?;
+    socket.set_recv_buffer_size(16 * 1024)?;
+    let stream = socket.connect(server.addr).await?;
+    let mut second = TestPeer::from_stream(stream);
+    second
+        .handshake(64702, Ipv4Addr::new(10, 0, 0, 2), 90)
+        .await?;
+    server.wait_for_sessions(2).await?;
+    // Sent over a couple of seconds, so that the server's export to the
+    // peer is well under way while the peer is still sending.
+    let sending = async {
+        for start in (0..ROUTES).step_by(100) {
+            second
+                .announce(Announcement::of(
+                    (start..start + 100).map(|i| nth(i).into()),
+                    &[64702, 64510],
+                ))
+                .await?;
+            if start % 2000 == 0 {
+                tokio::time::sleep(Duration::from_millis(40)).await;
+            }
+        }
+        Ok::<_, eyre::Report>(())
+    };
+    tokio::time::timeout(Duration::from_secs(30), sending)
+        .await
+        .expect("the server reads the peer's table though the peer reads nothing")?;
+    tokio::time::timeout(
+        Duration::from_secs(30),
+        server.wait_for_routes(2 * ROUTES as usize),
+    )
+    .await
+    .expect("every route is in the table")?;
+    Ok(())
+}