ipc: support long living sockets

src/ipc/server.rs

@@ -185,15 +185,23 @@ fn on_new_ipc_client(state: &mut State, stream: UnixStream) {
 
 async fn handle_client(ctx: ClientCtx, stream: Async<'static, UnixStream>) -> anyhow::Result<()> {
     let (read, mut write) = stream.split();
-    let mut buf = String::new();
+    let mut read = BufReader::new(read);
 
-    // Read a single line to allow extensibility in the future to keep reading.
-    BufReader::new(read)
-        .read_line(&mut buf)
-        .await
-        .context("error reading request")?;
+    loop {
+        // Don't keep buf around to avoid clients wasting RAM by filling it with bogus data.
+        let mut buf = Vec::new();
+        let res = read.read_until(b'\n', &mut buf).await;
+        match res {
+            Ok(0) => return Ok(()),
+            Ok(_) => (),
+            // Normal client disconnection.
+            Err(err) if err.kind() == io::ErrorKind::BrokenPipe => return Ok(()),
+            Err(err) => {
+                return Err(err).context("error reading request");
+            }
+        }
 
-    let request = serde_json::from_str(&buf)
+        let request = serde_json::from_slice(&buf)
             .context("error parsing request")
             .map_err(|err| err.to_string());
         let requested_error = matches!(request, Ok(Request::ReturnError));
@@ -210,7 +218,8 @@ async fn handle_client(ctx: ClientCtx, stream: Async<'static, UnixStream>) -> an
             }
         }
 
-    let mut buf = serde_json::to_vec(&reply).context("error formatting reply")?;
+        buf.clear();
+        serde_json::to_writer(&mut buf, &reply).context("error formatting reply")?;
         buf.push(b'\n');
         write.write_all(&buf).await.context("error writing reply")?;
 
@@ -252,9 +261,10 @@ async fn handle_client(ctx: ClientCtx, stream: Async<'static, UnixStream>) -> an
                 };
                 streams.push(sender);
             }
-    }
 
-    Ok(())
+            return Ok(());
+        }
+    }
 }
 
 async fn process(ctx: &ClientCtx, request: Request) -> Reply {