From 5847f6ee8044c70b23201ac5587ed59398659783 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Mon, 31 Aug 2026 23:19:25 +0000 Subject: [PATCH] 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 --- core/services/cluster/splice.go | 31 +++++++--- core/services/cluster/splice_test.go | 88 +++++++++++++++++++++++++++- 2 files changed, 109 insertions(+), 10 deletions(-) diff --git a/core/services/cluster/splice.go b/core/services/cluster/splice.go index ba5ac28af..5df97e890 100644 --- a/core/services/cluster/splice.go +++ b/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 } diff --git a/core/services/cluster/splice_test.go b/core/services/cluster/splice_test.go index f18618b6b..7bbe3246d 100644 --- a/core/services/cluster/splice_test.go +++ b/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)