diff --git a/apps/nesinit/README.md b/apps/nesinit/README.md
index ed63d7d6..0aec304e 100644
--- a/apps/nesinit/README.md
+++ b/apps/nesinit/README.md
@@ -65,6 +65,19 @@ Both directions. Inside the guest an envelope crosses a unix socket at
 That socket is a mechanism and expected to change; the envelope is the boundary
 and is not.
 
+Nothing is held for a workload that is not on the relay, and nothing waits on
+one that is slow to read. An envelope that arrives with nobody connected is
+dropped, and so is one that arrives faster than the workload reads: what
+crosses this layer is re-sent when it changes, so a queued copy is a stale copy
+— and the queue that would hold it is on the same loop that carries stop,
+shutdown and the workload's exit, none of which may wait behind it.
+
+A frame is capped at 64 KiB. The workload is on the other end of that socket
+and can write for as long as it likes without ever sending a newline; the
+process assembling it is the one the kernel has been told not to kill, so an
+unbounded buffer there comes out of everything else in the guest. Past the cap
+the connection is dropped and the relay waits for the next one.
+
 `body` is a string rather than nested JSON, deliberately. A document nesinit
 can index into is a document nesinit can grow to depend on, and then the layer
 is no longer opaque and the boundary it exists to draw is gone.
diff --git a/apps/nesinit/src/payload.rs b/apps/nesinit/src/payload.rs
index 1340f967..96b3a4ad 100644
--- a/apps/nesinit/src/payload.rs
+++ b/apps/nesinit/src/payload.rs
@@ -12,13 +12,22 @@ use std::io;
 use std::path::Path;
 
 use nesprotocol::lifecycle::{Payload, from_line, to_line};
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{UnixListener, UnixStream};
 use tokio::sync::mpsc::{Receiver, Sender};
 
 /// Where the workload finds the relay.
 pub const SOCKET: &str = "/nestri/payload.sock";
 
+/// The longest envelope this will assemble before giving up on the connection.
+///
+/// A cap rather than a preference. The workload is on the other end of this
+/// socket and can write for as long as it likes without ever sending a
+/// newline; without a limit, the process holding the buffer is the one process
+/// in the guest the kernel has been told not to kill, so the memory it takes
+/// comes out of everything else.
+const LONGEST_ENVELOPE: usize = 64 * 1024;
+
 /// The two directions, as the session sees them.
 pub struct Ports {
     /// Envelopes bound for the workload.
@@ -47,7 +56,25 @@ pub async fn serve(
     let listener = UnixListener::bind(path)?;
 
     loop {
-        let (stream, _) = listener.accept().await?;
+        let stream = tokio::select! {
+            accepted = listener.accept() => accepted?.0,
+            waiting = outbound.recv() => {
+                // Dropped rather than queued for whoever connects next. What
+                // crosses this layer is re-sent when it changes, so what a
+                // queue would hold is a stale copy, and holding it is also
+                // what would eventually block the session that fills it.
+                match waiting {
+                    Some(envelope) => {
+                        tracing::warn!(
+                            envelope = %envelope.summary(),
+                            "dropped an envelope: nothing is on the relay",
+                        );
+                        continue;
+                    }
+                    None => return Ok(()), // the session is over
+                }
+            }
+        };
         tracing::info!("the workload is on the relay");
         match relay(stream, &mut outbound, &inbound).await {
             Ok(()) => tracing::info!("the workload left the relay"),
@@ -62,19 +89,27 @@ async fn relay(
     inbound: &Sender<Payload>,
 ) -> io::Result<()> {
     let (reader, mut writer) = stream.into_split();
-    let mut lines = BufReader::new(reader).lines();
+    let mut reader = BufReader::new(reader);
+    let mut frame = Vec::new();
 
     loop {
         tokio::select! {
-            line = lines.next_line() => {
-                let Some(line) = line? else { return Ok(()) };
+            read = read_capped(&mut reader, &mut frame) => {
+                if !read? {
+                    return Ok(()); // the workload closed the socket
+                }
                 // Deliberately not `{:?}` on the error either: a decoder is
                 // entitled to quote what it could not read, and what it could
                 // not read is a body.
-                let Ok(payload) = from_line::<Payload>(&line) else {
-                    tracing::warn!(bytes = line.len(), "ignoring an envelope that would not decode");
+                let decoded = std::str::from_utf8(&frame)
+                    .ok()
+                    .and_then(|line| from_line::<Payload>(line).ok());
+                let Some(payload) = decoded else {
+                    tracing::warn!(bytes = frame.len(), "ignoring an envelope that would not decode");
+                    frame.clear();
                     continue;
                 };
+                frame.clear();
                 tracing::debug!(envelope = %payload.summary(), "relaying from the workload");
                 if inbound.send(payload).await.is_err() {
                     return Ok(()); // the session is over
@@ -91,10 +126,62 @@ async fn relay(
     }
 }
 
+/// Read one newline-terminated frame into `frame`, refusing to grow it past
+/// [`LONGEST_ENVELOPE`]. `false` at end of stream.
+///
+/// Cancel-safe, which it has to be to sit in a `select!`: bytes are copied out
+/// of the reader and consumed together, so a cancelled read leaves the partial
+/// frame in `frame` and the rest of it in the socket.
+async fn read_capped<R>(reader: &mut R, frame: &mut Vec<u8>) -> io::Result<bool>
+where
+    R: AsyncBufRead + Unpin,
+{
+    loop {
+        let consumed;
+        let complete;
+        {
+            let available = reader.fill_buf().await?;
+            if available.is_empty() {
+                return Ok(false);
+            }
+            match available.iter().position(|byte| *byte == b'\n') {
+                Some(end) => {
+                    within_cap(frame.len() + end)?;
+                    frame.extend_from_slice(&available[..end]);
+                    consumed = end + 1;
+                    complete = true;
+                }
+                None => {
+                    within_cap(frame.len() + available.len())?;
+                    frame.extend_from_slice(available);
+                    consumed = available.len();
+                    complete = false;
+                }
+            }
+        }
+        reader.consume(consumed);
+        if complete {
+            return Ok(true);
+        }
+    }
+}
+
+/// The error says how long, and nothing about what: what did not fit is a
+/// body, and a body is not logged even when it is malformed.
+fn within_cap(length: usize) -> io::Result<()> {
+    if length > LONGEST_ENVELOPE {
+        return Err(io::Error::other(format!(
+            "an envelope grew past {LONGEST_ENVELOPE} bytes without ending"
+        )));
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use std::sync::{Arc, Mutex};
+    use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf};
     use tokio::sync::mpsc;
 
     fn socket_path(name: &str) -> std::path::PathBuf {
@@ -128,22 +215,79 @@ mod tests {
         }
     }
 
+    /// A relay, and the two ends of it a test drives.
+    struct Relay {
+        path: std::path::PathBuf,
+        down: mpsc::Sender<Payload>,
+        up: mpsc::Receiver<Payload>,
+        server: tokio::task::JoinHandle<io::Result<()>>,
+    }
+
+    impl Relay {
+        fn start(name: &str) -> Self {
+            let path = socket_path(name);
+            let (down, outbound) = mpsc::channel(8);
+            let (inbound, up) = mpsc::channel(8);
+            let server = tokio::spawn({
+                let path = path.clone();
+                async move { serve(&path, outbound, inbound).await }
+            });
+            Self {
+                path,
+                down,
+                up,
+                server,
+            }
+        }
+
+        /// Dial the relay and wait until it is demonstrably carrying the
+        /// connection.
+        ///
+        /// `serve` drops what arrives while nothing is connected, so a test
+        /// that sends downward before the accept has completed is racing it. A
+        /// line upward is the barrier: it can only arrive once the relay is
+        /// carrying this connection.
+        async fn workload(&mut self) -> (BufReader<OwnedReadHalf>, OwnedWriteHalf) {
+            let (reader, mut writer) = self.dial().await.into_split();
+            writer
+                .write_all(to_line(&Payload::new("handshake", "")).unwrap().as_bytes())
+                .await
+                .unwrap();
+            assert_eq!(self.up.recv().await.unwrap().channel, "handshake");
+            (BufReader::new(reader), writer)
+        }
+
+        /// The listener may not be bound the instant the task is spawned.
+        async fn dial(&self) -> UnixStream {
+            for _ in 0..100 {
+                if let Ok(stream) = UnixStream::connect(&self.path).await {
+                    return stream;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+            }
+            panic!("the relay never came up on {}", self.path.display());
+        }
+    }
+
+    impl Drop for Relay {
+        fn drop(&mut self) {
+            self.server.abort();
+            let _ = std::fs::remove_file(&self.path);
+        }
+    }
+
     #[tokio::test]
     async fn an_envelope_crosses_in_both_directions_untouched() {
-        let path = socket_path("both-ways");
-        let (down_tx, down_rx) = mpsc::channel(4);
-        let (up_tx, mut up_rx) = mpsc::channel(4);
-        let server = tokio::spawn({
-            let path = path.clone();
-            async move { serve(&path, down_rx, up_tx).await }
-        });
-
-        let workload = connect(&path).await;
-        let (reader, mut writer) = workload.into_split();
-        let mut lines = BufReader::new(reader).lines();
+        let mut relay = Relay::start("both-ways");
+        let (reader, mut writer) = relay.workload().await;
+        let mut lines = reader.lines();
 
         let body = r#"{"looks":"structured"} and is not"#;
-        down_tx.send(Payload::new("identity", body)).await.unwrap();
+        relay
+            .down
+            .send(Payload::new("identity", body))
+            .await
+            .unwrap();
         let line = lines.next_line().await.unwrap().unwrap();
         let seen: Payload = from_line(&line).unwrap();
         assert_eq!(seen.body, body, "the body arrived changed");
@@ -151,11 +295,7 @@ mod tests {
 
         let back = to_line(&Payload::new("identity", "opaque back")).unwrap();
         writer.write_all(back.as_bytes()).await.unwrap();
-        let up = up_rx.recv().await.unwrap();
-        assert_eq!(up.body, "opaque back");
-
-        server.abort();
-        let _ = std::fs::remove_file(&path);
+        assert_eq!(relay.up.recv().await.unwrap().body, "opaque back");
     }
 
     #[tokio::test]
@@ -168,20 +308,13 @@ mod tests {
             .finish();
         let _log = tracing::subscriber::set_default(subscriber);
 
-        let path = socket_path("logging");
-        let (down_tx, down_rx) = mpsc::channel(4);
-        let (up_tx, mut up_rx) = mpsc::channel(4);
-        let server = tokio::spawn({
-            let path = path.clone();
-            async move { serve(&path, down_rx, up_tx).await }
-        });
-
-        let workload = connect(&path).await;
-        let (reader, mut writer) = workload.into_split();
-        let mut lines = BufReader::new(reader).lines();
+        let mut relay = Relay::start("logging");
+        let (reader, mut writer) = relay.workload().await;
+        let mut lines = reader.lines();
 
         let secret = "a-credential-nobody-should-read";
-        down_tx
+        relay
+            .down
             .send(Payload::new("identity", secret))
             .await
             .unwrap();
@@ -195,7 +328,7 @@ mod tests {
             )
             .await
             .unwrap();
-        up_rx.recv().await.unwrap();
+        relay.up.recv().await.unwrap();
 
         // An envelope that will not decode is the other way a body reaches a
         // log line — a decoder is entitled to quote what it could not read —
@@ -219,19 +352,91 @@ mod tests {
             logged.contains("bytes=62"),
             "the byte count is loggable:\n{logged}"
         );
-
-        server.abort();
-        let _ = std::fs::remove_file(&path);
     }
 
-    /// The listener may not be bound the instant the task is spawned.
-    async fn connect(path: &Path) -> UnixStream {
-        for _ in 0..100 {
-            if let Ok(stream) = UnixStream::connect(path).await {
-                return stream;
-            }
-            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
-        }
-        panic!("the relay never came up on {}", path.display());
+    #[tokio::test]
+    async fn an_envelope_sent_while_nobody_is_connected_is_dropped_rather_than_queued() {
+        let mut relay = Relay::start("nobody-home");
+
+        // Sent into a relay nothing is on. It is not held for whoever connects
+        // next: what crosses this layer is re-sent when it changes, so a held
+        // copy is a stale copy — and a queue that fills is what would
+        // eventually stall the session filling it.
+        relay
+            .down
+            .send(Payload::new("identity", "stale"))
+            .await
+            .unwrap();
+
+        let (reader, _writer) = relay.workload().await;
+        let mut lines = reader.lines();
+        relay
+            .down
+            .send(Payload::new("identity", "current"))
+            .await
+            .unwrap();
+
+        let line = lines.next_line().await.unwrap().unwrap();
+        let seen: Payload = from_line(&line).unwrap();
+        assert_eq!(
+            seen.body, "current",
+            "a stale envelope was delivered on connect"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_frame_that_never_ends_costs_the_connection_and_not_the_guest() {
+        use tokio::io::AsyncReadExt;
+
+        let mut relay = Relay::start("unbounded");
+        let (mut reader, mut writer) = relay.dial().await.into_split();
+
+        // Well past the cap, and not a newline in it. Without a limit the
+        // buffer holding this grows in the one process the kernel has been
+        // told not to kill, so what it takes comes out of everything else.
+        let flood = vec![b'a'; LONGEST_ENVELOPE + 4096];
+        // The write fails once the relay drops the connection, which is the
+        // outcome under test rather than a problem with it.
+        let _ = writer.write_all(&flood).await;
+
+        // The relay closes on us rather than keeping the buffer. Bounded,
+        // because the failure being guarded against is a relay that reads for
+        // as long as the workload writes.
+        let mut unread = Vec::new();
+        let closed = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            reader.read_to_end(&mut unread),
+        )
+        .await
+        .expect("the relay is still assembling a frame that never ends");
+        assert_eq!(
+            closed.unwrap(),
+            0,
+            "the relay answered a frame it should have refused"
+        );
+
+        // And it is there for whoever connects next.
+        let (reader, mut writer) =
+            tokio::time::timeout(std::time::Duration::from_secs(5), relay.workload())
+                .await
+                .expect("the relay never came back for the next workload");
+        let mut lines = reader.lines();
+        writer
+            .write_all(
+                to_line(&Payload::new("identity", "after the flood"))
+                    .unwrap()
+                    .as_bytes(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(relay.up.recv().await.unwrap().body, "after the flood");
+
+        relay
+            .down
+            .send(Payload::new("identity", "down"))
+            .await
+            .unwrap();
+        let line = lines.next_line().await.unwrap().unwrap();
+        assert_eq!(from_line::<Payload>(&line).unwrap().body, "down");
     }
 }
diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs
index 0bb302cb..78e0e28d 100644
--- a/apps/nesinit/src/session.rs
+++ b/apps/nesinit/src/session.rs
@@ -183,7 +183,7 @@ where
                     }
                 }
             }
-            HostToGuest::Payload { payload: envelope } => hand_over(payload, envelope).await,
+            HostToGuest::Payload { payload: envelope } => hand_over(payload, envelope),
             HostToGuest::Stop => workload.signal_stop(),
             HostToGuest::Shutdown => return Ok(Outcome::Shutdown),
         }
@@ -199,13 +199,24 @@ enum Event {
 
 /// Hand an envelope to the relay, and treat a relay that is not there as the
 /// caller's problem rather than a failure of this session.
-async fn hand_over(ports: &mut Ports, envelope: Payload) {
-    tracing::debug!(envelope = %envelope.summary(), "handing an envelope over");
-    if ports.to_workload.send(envelope).await.is_err() {
-        // The workload is not on the relay. Dropped rather than queued: what
-        // crosses here is re-sent when it changes, so a held copy is a stale
-        // copy.
-        tracing::warn!("dropped an envelope: nothing is on the relay");
+///
+/// It never waits. This loop also carries stop, shutdown and the workload's
+/// exit, and none of those may be held up by a workload that is slow to read
+/// its own mail — or by one that never connected at all. What crosses this
+/// layer is re-sent when it changes, so a dropped copy costs less than a
+/// stalled session.
+fn hand_over(ports: &mut Ports, envelope: Payload) {
+    use tokio::sync::mpsc::error::TrySendError;
+
+    let summary = envelope.summary();
+    match ports.to_workload.try_send(envelope) {
+        Ok(()) => tracing::debug!(envelope = %summary, "handed an envelope over"),
+        Err(TrySendError::Full(_)) => {
+            tracing::warn!(envelope = %summary, "dropped an envelope: the relay is behind")
+        }
+        Err(TrySendError::Closed(_)) => {
+            tracing::warn!(envelope = %summary, "dropped an envelope: the relay is gone")
+        }
     }
 }
 
@@ -655,4 +666,55 @@ mod tests {
 
         assert_eq!(session.await.unwrap(), Outcome::Shutdown);
     }
+
+    #[tokio::test]
+    async fn a_relay_that_is_not_draining_does_not_stall_the_session() {
+        // Bounded, because the failure is a session that stops rather than one
+        // that answers wrongly.
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            a_backed_up_relay().await
+        })
+        .await
+        .expect("the session stalled on the relay");
+    }
+
+    async fn a_backed_up_relay() {
+        let (guest, host) = tokio::io::duplex(4096);
+        let mut caller = Caller::new(host);
+        let (down_tx, down_rx) = mpsc::channel(1);
+        let (_up_tx, up_rx) = mpsc::channel::<Payload>(1);
+        // Held and never read: a workload that is slow to read its own mail,
+        // or one that connected and stopped.
+        let _backed_up = down_rx;
+
+        let session = tokio::spawn(async move {
+            let mut ports = Ports {
+                to_workload: down_tx,
+                from_workload: up_rx,
+            };
+            let mut workload = Double::exits_when_stopped(Exit::code(0));
+            run(guest, &mut workload, &mut ports).await.unwrap()
+        });
+
+        assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
+        for _ in 0..8 {
+            caller
+                .say(&HostToGuest::Payload {
+                    payload: Payload::new("identity", "backlog"),
+                })
+                .await;
+        }
+
+        // The lifecycle layer still moves: stop, shutdown and an exit are on
+        // this loop too, and none of them may wait on the relay.
+        caller
+            .say(&HostToGuest::Boot {
+                descriptor: Box::new(descriptor()),
+            })
+            .await;
+        caller.expect_started().await;
+        caller.say(&HostToGuest::Shutdown).await;
+
+        assert_eq!(session.await.unwrap(), Outcome::Shutdown);
+    }
 }
diff --git a/apps/nesinit/src/workload.rs b/apps/nesinit/src/workload.rs
index ef5d00f6..d9f7c3b9 100644
--- a/apps/nesinit/src/workload.rs
+++ b/apps/nesinit/src/workload.rs
@@ -192,11 +192,15 @@ impl Workload for Process {
 /// channel, so the guest still learns nothing about the filesystem it is being
 /// handed a piece of.
 fn mount_share(share: &Mount) -> Result<(), Failure> {
+    // Checked before anything is created: a descriptor this component cannot
+    // act on should leave no directory behind to confuse whoever reads the
+    // failure.
+    let (source, target, flags) = options(share)?;
+
     // The mount point may not exist yet: a share can land anywhere the
     // descriptor names, including a directory no image created.
     std::fs::create_dir_all(&share.at).map_err(|error| failed(share, error))?;
 
-    let (source, target, flags) = options(share);
     // SAFETY: mount takes two paths, a filesystem name and a flag word, all
     // of which outlive the call, and no options string.
     let mounted = unsafe {
@@ -220,7 +224,7 @@ const FSTYPE: &std::ffi::CStr = c"virtiofs";
 
 /// What the mount call is given, split out because this is the part worth
 /// asserting: mounting itself needs privileges a test does not have.
-fn options(share: &Mount) -> (CString, CString, libc::c_ulong) {
+fn options(share: &Mount) -> Result<(CString, CString, libc::c_ulong), Failure> {
     // nosuid and nodev on every share, whether or not it is writable. A share
     // is data handed to the guest; a setuid binary or a device node appearing
     // in one is not something a workload should be able to use, and no
@@ -229,11 +233,23 @@ fn options(share: &Mount) -> (CString, CString, libc::c_ulong) {
     if share.ro {
         flags |= libc::MS_RDONLY;
     }
-    // Interior nul bytes are the caller's mistake; a path with one cannot be
-    // mounted under any flags.
-    let source = CString::new(share.tag.as_str()).unwrap_or_default();
-    let target = CString::new(share.at.as_str()).unwrap_or_default();
-    (source, target, flags)
+    // A nul byte inside a tag or a path is a descriptor that cannot be carried
+    // out under any flags. Refused by name rather than silently emptied: an
+    // empty source turns up later as a mount failure about something else
+    // entirely, which is the wrong thing to go and look at.
+    let source = CString::new(share.tag.as_str()).map_err(|_| {
+        Failure::new(format!(
+            "the share tag contains a nul byte: {:?}",
+            share.tag
+        ))
+    })?;
+    let target = CString::new(share.at.as_str()).map_err(|_| {
+        Failure::new(format!(
+            "the mount point contains a nul byte: {:?}",
+            share.at
+        ))
+    })?;
+    Ok((source, target, flags))
 }
 
 /// A failure names the path, which is what makes it actionable: a permission
@@ -257,7 +273,7 @@ mod tests {
 
     #[test]
     fn a_writable_share_is_still_mounted_without_devices_or_setuid() {
-        let (source, target, flags) = options(&share(false));
+        let (source, target, flags) = options(&share(false)).unwrap();
         assert_eq!(
             source.to_str().unwrap(),
             "user",
@@ -271,10 +287,29 @@ mod tests {
 
     #[test]
     fn a_read_only_share_is_mounted_read_only() {
-        let (_, _, flags) = options(&share(true));
+        let (_, _, flags) = options(&share(true)).unwrap();
         assert_eq!(flags & libc::MS_RDONLY, libc::MS_RDONLY);
     }
 
+    #[test]
+    fn a_descriptor_with_a_nul_byte_in_it_is_refused_by_name() {
+        let tagged = Mount {
+            tag: "us\0er".into(),
+            at: "/mnt/user".into(),
+            ro: false,
+        };
+        let failure = options(&tagged).expect_err("an empty source would have been mounted");
+        assert!(failure.reason.contains("tag"), "{}", failure.reason);
+
+        let placed = Mount {
+            tag: "user".into(),
+            at: "/mnt/us\0er".into(),
+            ro: false,
+        };
+        let failure = options(&placed).expect_err("an empty target would have been mounted");
+        assert!(failure.reason.contains("mount point"), "{}", failure.reason);
+    }
+
     #[test]
     fn a_failure_names_the_path_it_happened_on() {
         let failure = failed(&share(false), io::Error::from_raw_os_error(libc::EACCES));