bound the request line on the host admin and in-agent sockets

hive-c0re's admin socket and hive-agent's in-agent socket (mode 0666) read
the request with an unbounded read_line: no length cap and no deadline, so
a peer that never sends a newline grows a buffer and pins a task forever.

The bounded reader (16 MiB cap) moves from hive-c0re's socket_server into
hive-sock-client, the one crate both daemons already share or can take, and
gains optional deadlines for the first byte and for the rest of the line.
Both sockets now use it and answer an over-long or stalled line with the
same error reply the bounded sockets give, then close. The admin socket
bounds only the started line (clients may idle between requests); the
in-agent socket is single-shot and bounds silence too. The per-agent
sockets keep their behaviour (no deadline).

hive-c0re/src/server.rs

@@ -4,7 +4,8 @@ use std::sync::Arc;
 use anyhow::{Context, Result};
 use hive_host_sock::{HostRequest, HostResponse, LifecycleScope};
 use hive_priv_sock::{InfraAction, InfraContainer};
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use hive_sock_client::{LineDeadlines, REQUEST_LINE_DEADLINE, RequestLine, read_request_line};
+use tokio::io::{AsyncWriteExt, BufReader};
 use tokio::net::{UnixListener, UnixStream};
 
 use crate::actions;
@@ -53,17 +54,35 @@ pub async fn serve(socket: &Path, coord: Arc<Coordinator>) -> Result<()> {
 }
 
 async fn handle(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
+    handle_with(stream, coord, ADMIN_LINE_DEADLINES).await
+}
+
+/// A client may idle between requests on one connection, so only the time to
+/// finish a started line is bounded.
+const ADMIN_LINE_DEADLINES: LineDeadlines = LineDeadlines {
+    idle: None,
+    line: Some(REQUEST_LINE_DEADLINE),
+};
+
+async fn handle_with(
+    stream: UnixStream,
+    coord: Arc<Coordinator>,
+    deadlines: LineDeadlines,
+) -> Result<()> {
     let (read, mut write) = stream.into_split();
     let mut reader = BufReader::new(read);
-    let mut line = String::new();
+    let mut line = Vec::new();
 
     loop {
-        line.clear();
-        let n = reader.read_line(&mut line).await?;
-        if n == 0 {
+        let outcome = read_request_line(&mut reader, &mut line, deadlines).await?;
+        if outcome == RequestLine::Eof {
             return Ok(());
         }
-        let req = match serde_json::from_str::<HostRequest>(line.trim()) {
+        if let Some(message) = outcome.refusal() {
+            write_response(&mut write, &HostResponse::error(message)).await?;
+            return Ok(());
+        }
+        let req = match serde_json::from_slice::<HostRequest>(line.trim_ascii()) {
             Ok(req) => req,
             Err(e) => {
                 write_response(
@@ -828,3 +847,73 @@ fn finish_lifecycle(ok_items: Vec<String>, errors: &[String]) -> HostResponse {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use hive_sock_client::REQUEST_LINE_MAX_BYTES;
+    use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
+
+    use super::*;
+
+    /// A real `Coordinator` over a throwaway sqlite dir; `handle_with` takes
+    /// an `Arc<Coordinator>` and `open` touches nothing outside the db path.
+    fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let coord = Coordinator::open(
+            &dir.path().join("broker.sqlite"),
+            crate::coordinator::HiveEnv::default(),
+            crate::hive_stats::PriceTable::new(),
+            1,
+        )
+        .expect("open coordinator");
+        (dir, Arc::new(coord))
+    }
+
+    async fn reply_error(client: &mut tokio::net::unix::OwnedReadHalf) -> String {
+        let mut reply = String::new();
+        BufReader::new(client)
+            .read_line(&mut reply)
+            .await
+            .expect("reply");
+        let resp: HostResponse = serde_json::from_str(&reply).expect("response json");
+        resp.error.expect("error reply")
+    }
+
+    #[tokio::test]
+    async fn over_long_request_gets_an_error_and_the_connection_closes() {
+        let (_dir, coord) = coordinator();
+        let (client, server) = UnixStream::pair().expect("socketpair");
+        let handler = tokio::spawn(handle_with(server, coord, ADMIN_LINE_DEADLINES));
+        let (mut read, mut write) = client.into_split();
+        let writer = tokio::spawn(async move {
+            let chunk = vec![b'x'; 64 * 1024];
+            for _ in 0..2 * REQUEST_LINE_MAX_BYTES / chunk.len() {
+                if write.write_all(&chunk).await.is_err() {
+                    return;
+                }
+            }
+        });
+        let message = reply_error(&mut read).await;
+        assert!(message.starts_with("request too long"), "{message}");
+        handler.await.expect("join").expect("clean close");
+        writer.await.expect("join");
+    }
+
+    #[tokio::test]
+    async fn stalled_request_gets_an_error_and_the_connection_closes() {
+        let (_dir, coord) = coordinator();
+        let (client, server) = UnixStream::pair().expect("socketpair");
+        let deadlines = LineDeadlines {
+            idle: None,
+            line: Some(Duration::from_millis(50)),
+        };
+        let handler = tokio::spawn(handle_with(server, coord, deadlines));
+        let (mut read, mut write) = client.into_split();
+        write.write_all(b"{\"partial\"").await.expect("write");
+        let message = reply_error(&mut read).await;
+        assert!(message.starts_with("request timed out"), "{message}");
+        handler.await.expect("join").expect("clean close");
+    }
+}