fix(cluster): stop reading a bare EOF as a clean ending

A dead yamux session does not always arrive wrapped. Session.close publishes
shutdownErr and closes shutdownCh before it force-closes the streams, so a
Write or Close landing in that window gets the raw cause back instead
(session.go:507-510, 528-533), and for a peer that vanished the raw cause is a
bare io.EOF. The generic io.EOF clause then reported the dead session as a
clean completion.

Remove the clause. A clean read-side EOF never reached it anyway: io.Copy
consumes that and reports nil, and neither *yamux.Stream nor *net.TCPConn
takes a WriteTo/ReadFrom path that would hand one back. Every existing spec
still passes, the io.EOF entry in the normal-termination table included, which
is what showed the branch was dead for legitimate endings and live only for
the bug.

Add a spec driving a real yamux session end to end. Every mux shape until now
was a synthesized error, which is exactly why a race inside the real library
stayed invisible.

Assisted-by: Claude Opus 5 [claude-code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

core/services/cluster/splice.go

@@ -69,15 +69,22 @@ func copyStream(dst io.Writer, src io.Reader) error {
 }
 
 // normalizeStreamErr drops the endings that mean the conversation is over
-// rather than broken. io.EOF is the clean end of a stream, net.ErrClosed is
-// what a socket reports once it or its peer has been closed, and
-// io.ErrClosedPipe is the same condition on an in-memory pipe.
+// rather than broken: net.ErrClosed is what a socket reports once it or its
+// peer has been closed, and io.ErrClosedPipe is the same condition on an
+// in-memory pipe.
+//
+// io.EOF is deliberately absent. A clean read-side EOF never gets this far,
+// because io.Copy consumes it and reports nil, and neither *yamux.Stream nor
+// *net.TCPConn takes a WriteTo/ReadFrom path that would hand one back. So a
+// bare io.EOF arriving here came from a failing Write or Close, where it means
+// the peer is gone, and yamux produces exactly that when a Write races its
+// session's shutdown (see isMuxSessionFailure).
 //
 // The mux checks run first, and that ordering is load-bearing: a dying yamux
-// session hands every live stream its own cause wrapped up (session.go:330),
-// and that cause is routinely io.EOF or a closed-socket error, so consulting
-// the generic endings first would report a peer that vanished mid-request as a
-// clean completion.
+// session usually hands every live stream its own cause wrapped up
+// (session.go:330), and that cause is routinely a closed-socket error, so
+// consulting the generic endings first would report a peer that vanished
+// mid-request as a clean completion.
 func normalizeStreamErr(err error) error {
 	if err == nil {
 		return nil
@@ -88,8 +95,7 @@ func normalizeStreamErr(err error) error {
 	if isMuxStreamTeardown(err) {
 		return nil
 	}
-	if errors.Is(err, io.EOF) ||
-		errors.Is(err, net.ErrClosed) ||
+	if errors.Is(err, net.ErrClosed) ||
 		errors.Is(err, io.ErrClosedPipe) {
 		return nil
 	}
@@ -122,6 +128,13 @@ func isMuxSessionFailure(err error) bool {
 	// the cause, so the bare sentinel means this stream was reset and a
 	// wrapped one means the session died under it. Identity is what separates
 	// them; errors.Is cannot.
+	//
+	// Wrapped is not the only way a dead session shows up, though. close()
+	// publishes shutdownErr and closes shutdownCh before it force-closes the
+	// streams, so a Write or Close landing in that window gets the raw cause
+	// back instead (session.go:507-510, 528-533). That form is unrecognisable
+	// as yamux at all, which is why normalizeStreamErr no longer forgives a
+	// bare io.EOF: for a peer that vanished, the raw cause is precisely io.EOF.
 	return errors.Is(err, yamux.ErrStreamReset) && err != yamux.ErrStreamReset
 }
 

core/services/cluster/splice_test.go

@@ -1,6 +1,7 @@
 package cluster_test
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -206,6 +207,23 @@ var _ = Describe("Splice", func() {
 			Entry("an internal-error go-away", &yamux.GoAwayError{Remote: true, ErrorCode: 2}),
 		)
 
+		// A bare io.EOF can only reach Splice from a failing Write. io.Copy
+		// never surfaces a clean read-side EOF, and yamux hands out the raw
+		// cause rather than the wrapped one when a Write or Close races
+		// Session.close's shutdown window (session.go:507-510), so for a
+		// vanished peer this IS the dead session, arriving unwrapped.
+		It("reports a write that fails with a bare EOF", func() {
+			sink := &scriptedStream{writeErr: io.EOF}
+			source := &scriptedStream{feeds: true}
+
+			done := make(chan error, 1)
+			go func() { done <- cluster.Splice(sink, source) }()
+
+			var err error
+			Eventually(done, "5s").Should(Receive(&err))
+			Expect(err).To(MatchError(io.EOF))
+		})
+
 		// The distinction the classifier turns on, in one spec: yamux uses the
 		// same sentinel for "this stream was reset", which Splice provokes
 		// itself and must stay quiet about, and as the head of the wrapped
@@ -254,6 +272,57 @@ var _ = Describe("Splice", func() {
 			Eventually(done, "5s").Should(Receive(BeNil()))
 		})
 
+		// Everything above feeds Splice a synthesized error. This one drives a
+		// real yamux session, because the shapes a live library produces are
+		// not always the ones its source suggests: the bug this spec was added
+		// alongside was a race inside Session.close that no synthesized error
+		// could show. It asserts only that a dead session is reported, not how
+		// it is spelled, since which of the two forms arrives is a race.
+		It("reports a real yamux session dying under a live stream", func() {
+			clientConn, serverConn := net.Pipe()
+			client, err := yamux.Client(clientConn, nil, nil)
+			Expect(err).ToNot(HaveOccurred())
+			server, err := yamux.Server(serverConn, nil, nil)
+			Expect(err).ToNot(HaveOccurred())
+			DeferCleanup(func() {
+				_ = client.Close()
+				_ = server.Close()
+			})
+
+			accepted := make(chan *yamux.Stream, 1)
+			go func() {
+				defer GinkgoRecover()
+				far, err := server.AcceptStream()
+				if err != nil {
+					close(accepted)
+					return
+				}
+				accepted <- far
+			}()
+
+			stream, err := client.OpenStream(context.Background())
+			Expect(err).ToNot(HaveOccurred())
+			// Push a byte so the stream is established on both sides before
+			// the session is killed.
+			_, err = stream.Write([]byte("x"))
+			Expect(err).ToNot(HaveOccurred())
+			var far *yamux.Stream
+			Eventually(accepted, "10s").Should(Receive(&far))
+			_, err = far.Read(make([]byte, 1))
+			Expect(err).ToNot(HaveOccurred())
+
+			done := make(chan error, 1)
+			go func() { done <- cluster.Splice(stream, &scriptedStream{}) }()
+
+			// The peer's process disappears: the connection carrying the
+			// session goes away, which kills every stream riding on it.
+			Expect(serverConn.Close()).To(Succeed())
+
+			var spliceErr error
+			Eventually(done, "10s").Should(Receive(&spliceErr))
+			Expect(spliceErr).To(HaveOccurred())
+		})
+
 		// The anti-leak guarantee, which the pipe specs cannot see because
 		// their parked copy is released too quickly to catch Splice in the
 		// act. Waking a copy is asynchronous on a real stream (yamux's Close
@@ -281,7 +350,11 @@ var _ = Describe("Splice", func() {
 // parks in Read until Close, standing in for an idle half of a live stream.
 type scriptedStream struct {
 	readErr  error
+	writeErr error
 	closeErr error
+	// feeds makes Read produce bytes instead of parking, so a spec can keep a
+	// direction copying until its destination fails.
+	feeds bool
 	// holdReadPastClose keeps a parked Read blocked until release is called,
 	// standing in for the gap between a Close waking a reader and that reader
 	// running. Without it, Close releases the Read as a real stream does.
@@ -307,11 +380,24 @@ func (s *scriptedStream) Read(p []byte) (int, error) {
 	if s.readErr != nil {
 		return 0, s.readErr
 	}
+	if s.feeds {
+		select {
+		case <-s.gate():
+			return 0, io.EOF
+		default:
+			return len(p), nil
+		}
+	}
 	<-s.gate()
 	return 0, io.EOF
 }
 
-func (s *scriptedStream) Write(p []byte) (int, error) { return len(p), nil }
+func (s *scriptedStream) Write(p []byte) (int, error) {
+	if s.writeErr != nil {
+		return 0, s.writeErr
+	}
+	return len(p), nil
+}
 
 func (s *scriptedStream) Close() error {
 	s.closeN.Add(1)