Verify streamed replay and release disconnected web clients

flake.nix

@@ -135,9 +135,10 @@
           installPhaseCommand = ''
             mkdir -p $out
             cp -a target/userland/bundle/. $out/
-            test -f $out/bin/chat.wasm
-            test -f $out/etc/systemd/system/chat.service
-            test -f $out/etc/systemd/system/multi-user.target.wants/chat.service
+            test -f $out/bin/web.wasm
+            test -f $out/srv/www/index.html
+            test -f $out/etc/systemd/system/web.service
+            test -f $out/etc/systemd/system/multi-user.target.wants/web.service
           '';
           doInstallCargoArtifacts = false;
           # The upstream browser fixture links to an optional, unbuilt artifact.

host/tests/it/chat.rs

@@ -283,6 +283,24 @@ async fn seeded_chat_case(open_connection: bool) -> anyhow::Result<()> {
             anyhow::ensure!(std::time::Instant::now() < deadline, "Reply did not appear");
             tokio::task::yield_now().await;
         }
+        let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port)).await?;
+        stream.write_all(format!("GET /api/harness/conversations/{root}/stream?after=0 HTTP/1.1\r\nHost: localhost:{port}\r\nConnection: close\r\n\r\n").as_bytes()).await?;
+        let mut streamed = Vec::new();
+        tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                let mut chunk = [0; 4096];
+                let count = stream.read(&mut chunk).await?;
+                anyhow::ensure!(count > 0, "Stream closed before replay");
+                streamed.extend_from_slice(&chunk[..count]);
+                if String::from_utf8_lossy(&streamed).contains("Chat test reply") {
+                    break;
+                }
+            }
+            Ok::<_, anyhow::Error>(())
+        })
+        .await??;
+        assert!(String::from_utf8_lossy(&streamed).contains("application/x-ndjson"));
+        drop(stream);
         let (status, body) = request(port, "/api/harness/credential-requests", None).await?;
         assert_eq!(status, 200);
         let value: serde_json::Value = serde_json::from_slice(&body)?;

host/tests/it/p1.rs

@@ -13,8 +13,8 @@ fn bundle() -> anyhow::Result<PathBuf> {
         || !bundle.join("etc/sausage/config.yaml").exists()
         || std::fs::read_to_string(&stamp).ok().as_deref() != Some(key.as_str())
     {
-        if stamp.exists() {
-            std::fs::remove_file(&stamp)?;
+        if bundle.exists() {
+            std::fs::remove_dir_all(&bundle)?;
         }
         assert!(
             std::process::Command::new("nu")

host/tests/it/p2.rs

@@ -390,7 +390,7 @@ async fn request_approve_fetch_and_ssh_push() -> Result<()> {
     ensure!(
         browser
             .raw("GET", "/register", Value::Null)?
-            .contains("/app.js"),
+            .contains("src=\"app.js\""),
         "registration page missing"
     );
     let start: Challenge<CreationChallengeResponse> = serde_json::from_value(browser.json(

userland/web/src/main.rs

@@ -54,16 +54,34 @@ fn write(stdin: &InputStream, stream: &OutputStream, mut bytes: &[u8]) -> io::Re
         }
     }
 }
+fn wait_response(stdin: &InputStream, event: &Pollable, client: &InputStream) -> io::Result<()> {
+    let stop = stdin.subscribe();
+    let disconnect = client.subscribe();
+    loop {
+        let ready = poll(&[&stop, event, &disconnect]);
+        match stdin.read(65536) {
+            Err(StreamError::Closed) => std::process::exit(0),
+            Err(e) => return Err(error(e)),
+            Ok(_) => {}
+        }
+        if ready.contains(&2) {
+            client.read(65536).map_err(error)?;
+        }
+        if ready.contains(&1) {
+            return Ok(());
+        }
+    }
+}
 fn proxy(
     stdin: &InputStream,
+    client: &InputStream,
     method: &str,
     path: &str,
-    content_type: Option<&str>,
-    origin: Option<&str>,
-    host: Option<&str>,
+    headers: (Option<&str>, Option<&str>, Option<&str>),
     body: &[u8],
     output: &OutputStream,
 ) -> io::Result<()> {
+    let (content_type, origin, host) = headers;
     let headers = Fields::new();
     headers
         .append("content-length", &body.len().to_string().into_bytes())
@@ -102,7 +120,7 @@ fn proxy(
         write(stdin, &stream, body)?;
     }
     OutgoingBody::finish(outgoing, None).map_err(error)?;
-    wait(stdin, &future.subscribe())?;
+    wait_response(stdin, &future.subscribe(), client)?;
     let response = future
         .get()
         .ok_or_else(|| error("missing HTTP response"))?
@@ -119,7 +137,7 @@ fn proxy(
     let stream = incoming.stream().map_err(error)?;
     write(stdin, output, format!("HTTP/1.1 {status} Response\r\nContent-Type: {content_type}\r\nConnection: close\r\n\r\n").as_bytes())?;
     loop {
-        wait(stdin, &stream.subscribe())?;
+        wait_response(stdin, &stream.subscribe(), client)?;
         match stream.read(65536) {
             Ok(chunk) => write(stdin, output, &chunk)?,
             Err(StreamError::Closed) => break,
@@ -165,11 +183,10 @@ fn serve(
     if let Some(path) = path.strip_prefix("/api/") {
         return proxy(
             stdin,
+            input,
             method,
             path,
-            header("content-type"),
-            header("origin"),
-            header("host"),
+            (header("content-type"), header("origin"), header("host")),
             &bytes[end..end + length],
             output,
         );