fix(cluster): express the splice policy once, and pin the relay's budgets

Review round 1 on task 5. Eight non-blocking items, all addressed.

The classifier read Remote in two predicates with a report-by-default
fallthrough behind them, so reverting either read left the whole suite green:
the error reached the same answer down the other path. A correctness argument
that rests on mutation evidence cannot afford a shape that cannot be mutated in
pieces, so the two predicates collapse into one muxVerdict deciding each error
type once. Falsifying either Remote read now reddens exactly one spec.

Three claims the comments made loudly and nothing tested:

  - clearing the header read deadline before the splice. Deleting the clear
    left all 49 focused specs green, while in production it is the difference
    between a relayed response that streams for an hour and one that dies after
    fifteen seconds of quiet;
  - the open budget bounding the open and nothing after it;
  - closing the worker-side stream when the acceptance reply cannot be
    written, which leaks one stream on the worker per failure.

All three are pinned now. The first two share a spec that sets both budgets to
50ms and then watches the conversation outlive them by ten times, which is an
assertion about an event that must not happen and so is the one wait a channel
cannot replace. The third drives the relay with a peer stream that delivers a
request and then fails every write, because no pair of live yamux sessions can
be made to fail that write on cue.

The disjoint-vocabulary argument was specced for the accepted frame only. Both
refusal directions are covered now, and asserted as "not one of the other hop's
sentinels" rather than merely "an error", since reading a relay refusal with the
tunnel's reader always errors and the question is whether it errors as the wrong
thing.

The open budget stays non-configurable, and says so: the number that matters is
how long the original client will wait, which is not known on this side and is
not something a deployment-wide constant can stand in for. The honest fix is the
caller's remaining budget travelling in the request frame, which belongs to the
dialler that has the budget.

Two comments corrected: nothing deadlines the peer stream after the clear, so
the tunnelled protocol's own deadlines cannot be what justifies clearing it; and
the membership sweep deletes departed replicas but reports only how many, so
identifying them is work that would have to be done, not knowledge waiting to be
plumbed. Recorded at muxVerdict: a remote RST that does not ride a
typeWindowUpdate frame yields the bare sentinel and is still silenced, which is
unreachable between two go-yamux peers but keeps the new rule from reading as
unconditional.

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

core/services/cluster/peerlink.go

@@ -233,10 +233,10 @@ func (p *PeerPool) Open(ctx context.Context, peerID string) (net.Conn, error) {
 // has left the deployment but is still listening keeps a live WebSocket and the
 // two yamux loop goroutines behind it for as long as this process runs; a peer
 // that is genuinely gone is reclaimed by the 30s keepalive default, so the real
-// exposure is narrow. There is no Forget: the membership sweep does know which
-// replicas have left, but nothing plumbs that knowledge to this pool, so an
-// entry for a departed peer outlives it and only the keepalive reclaims what it
-// holds.
+// exposure is narrow. There is no Forget: the membership sweep DELETES departed
+// replicas but reports only how many, so which ones they were would have to be
+// surfaced before anything could be plumbed here. Until it is, an entry for a
+// departed peer outlives it and only the keepalive reclaims what it holds.
 func (p *PeerPool) link(peerID string) (*peerLink, error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()

core/services/cluster/relay.go

@@ -178,6 +178,15 @@ const (
 	// which is the only reason there is one here. Without the bound, a worker
 	// that has stopped accepting would turn a refusable condition into a parked
 	// peer, which is the one outcome this path exists to avoid.
+	//
+	// It is deliberately NOT configurable. No operator has the information to
+	// set it: the number that matters is how long the ORIGINAL client is
+	// willing to wait, which is not known on this side of the link and is not
+	// something a deployment-wide constant can stand in for. The honest fix is
+	// the caller's remaining budget travelling in the relay request frame, and
+	// that belongs to the dialler that has the budget. Until then this is a
+	// backstop against parking, generous on purpose, because refusing healthy
+	// traffic costs more than waiting.
 	relayOpenTimeout = 15 * time.Second
 )
 
@@ -271,10 +280,14 @@ func (r *Relay) accept(peerID string, stream net.Conn) (net.Conn, bool) {
 		return nil, false
 	}
 
-	// Cleared before the open rather than after the reply: everything past this
-	// frame belongs to the worker tunnel's conversation, which brings its own
-	// deadlines, and one left armed here would abort a long inference stream in
-	// the middle.
+	// Cleared before the open rather than after the reply, and NOTHING arms
+	// another deadline on this stream afterwards. That is the intent rather
+	// than an omission: what follows is a relayed request whose length is the
+	// caller's business, and a header deadline left armed here would abort a
+	// long inference stream after any quiet moment in the middle of it. What
+	// still bounds the conversation is the peer link's own keepalive, which
+	// kills the session under it when the far side stops answering, and
+	// whatever deadline the original client is holding.
 	if err := stream.SetReadDeadline(time.Time{}); err != nil {
 		r.refuse(peerID, stream, fmt.Errorf("%w: clearing the request deadline: %v", ErrRelayUnavailable, err))
 		return nil, false

core/services/cluster/relay_internal_test.go

@@ -3,9 +3,16 @@
 package cluster
 
 import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
 	"net"
+	"sync"
 	"time"
 
+	"github.com/mudler/LocalAI/core/services/testutil"
+
 	"github.com/libp2p/go-yamux/v5"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -61,3 +68,241 @@ var _ = Describe("A peer stream that never says what it wants", func() {
 		Eventually(ends, "10s").Should(Receive(HaveOccurred()))
 	})
 })
+
+// backloggedPair returns a peer/relay session pair whose SYN backlog is one
+// stream deep, so a single un-accepted open fills it and the next one parks.
+// yamux's default is 256 (mux.go, DefaultConfig), and filling that from a spec
+// would mean 256 real opens to prove one property.
+func backloggedPair(backlog int) (dialled, accepted *yamux.Session) {
+	GinkgoHelper()
+	cfg := yamux.DefaultConfig()
+	cfg.AcceptBacklog = backlog
+	a, b := net.Pipe()
+	var err error
+	accepted, err = yamux.Server(a, cfg, nil)
+	Expect(err).ToNot(HaveOccurred())
+	dialled, err = yamux.Client(b, cfg, nil)
+	Expect(err).ToNot(HaveOccurred())
+	DeferCleanup(func() {
+		_ = dialled.Close()
+		_ = accepted.Close()
+	})
+	return dialled, accepted
+}
+
+// unwritableStream is a peer stream that delivers one relay request and then
+// fails every write. It stands in for a peer that vanished between opening the
+// stream and hearing the answer, which is the only way the acceptance reply
+// fails, and which no pair of live yamux sessions can be made to do on cue.
+type unwritableStream struct {
+	net.Conn
+	request  []byte
+	read     int
+	closed   chan struct{}
+	closeOne sync.Once
+}
+
+func newUnwritableStream(nodeID string) *unwritableStream {
+	GinkgoHelper()
+	frame := &bytes.Buffer{}
+	Expect(WriteRelayRequest(frame, nodeID)).To(Succeed())
+	return &unwritableStream{request: frame.Bytes(), closed: make(chan struct{})}
+}
+
+func (s *unwritableStream) Read(p []byte) (int, error) {
+	if s.read >= len(s.request) {
+		// Never EOF: an EOF here would end the relay for a reason other than
+		// the failed write, and the spec would pass without exercising it.
+		<-s.closed
+		return 0, io.EOF
+	}
+	n := copy(p, s.request[s.read:])
+	s.read += n
+	return n, nil
+}
+
+func (s *unwritableStream) Write([]byte) (int, error) { return 0, errors.New("peer went away") }
+
+func (s *unwritableStream) Close() error {
+	s.closeOne.Do(func() { close(s.closed) })
+	return nil
+}
+
+func (s *unwritableStream) SetReadDeadline(time.Time) error { return nil }
+
+var _ = Describe("The relay's own budgets", func() {
+	var (
+		reg *Registry
+		tun *TunnelRegistry
+		ctx context.Context
+	)
+
+	BeforeEach(func() {
+		ctx = context.Background()
+		db := testutil.SetupTestDB()
+		Expect(Migrate(ctx, db)).To(Succeed())
+		reg = NewRegistry(db)
+		Expect(reg.Register(ctx, "me", "10.0.0.1:8080", "v1")).To(Succeed())
+		tun = NewTunnelRegistry(reg, "me")
+	})
+
+	It("stops bounding the stream once the relay hands it over", func() {
+		// Both budgets are set to 50ms here and both are deliberately shorter
+		// than the window this spec then watches. A header deadline left armed
+		// past acceptance, or an open budget applied to the stream it produced,
+		// would abort a relayed inference after 50ms of quiet, which in
+		// production is the difference between a response that streams for an
+		// hour and one that dies mid-token.
+		relay := newRelay(tun, 50*time.Millisecond, 50*time.Millisecond)
+		store := NewSessionStore(relay.Stream)
+		DeferCleanup(store.CloseAll)
+		peer, accepted := backloggedPair(256)
+		store.Accept("peer-1", accepted)
+
+		worker, frontend := backloggedPair(256)
+		_, err := tun.Attach(ctx, "w1", frontend)
+		Expect(err).ToNot(HaveOccurred())
+
+		workerSide := make(chan net.Conn, 1)
+		go func() {
+			defer GinkgoRecover()
+			stream, err := worker.AcceptStream()
+			if err != nil {
+				return
+			}
+			workerSide <- stream
+		}()
+
+		stream, err := peer.OpenStream(ctx)
+		Expect(err).ToNot(HaveOccurred())
+		DeferCleanup(func() { _ = stream.Close() })
+		Expect(WriteRelayRequest(stream, "w1")).To(Succeed())
+
+		replies := make(chan error, 1)
+		go func() {
+			defer GinkgoRecover()
+			replies <- ReadRelayReply(stream)
+		}()
+		Eventually(replies, "10s").Should(Receive(BeNil()))
+
+		var served net.Conn
+		Eventually(workerSide, "10s").Should(Receive(&served))
+
+		// One reader for both questions, so that watching for a teardown does
+		// not eat the bytes the second half of the spec is waiting for.
+		data := make(chan []byte, 4)
+		ended := make(chan error, 1)
+		go func() {
+			defer GinkgoRecover()
+			buf := make([]byte, 64)
+			for {
+				n, err := served.Read(buf)
+				if n > 0 {
+					data <- append([]byte(nil), buf[:n]...)
+				}
+				if err != nil {
+					ended <- err
+					return
+				}
+			}
+		}()
+
+		// An assertion about an event that must NOT happen, which is the one
+		// kind a channel cannot replace: a torn-down splice ends the worker's
+		// side, and there is no event for "still alive". The window is ten
+		// times the budgets it is watching.
+		Consistently(ended, "500ms", "50ms").ShouldNot(Receive(),
+			"the relay tore the stream down on a budget that should have stopped applying at acceptance")
+
+		// And it is not merely un-torn-down: it still carries bytes, long after
+		// both budgets would have expired.
+		_, err = stream.Write([]byte("late"))
+		Expect(err).ToNot(HaveOccurred())
+		Eventually(data, "10s").Should(Receive(Equal([]byte("late"))))
+	})
+
+	It("refuses rather than parking when the worker's tunnel will not take another stream", func() {
+		// The open budget exists because yamux BLOCKS an open once the accept
+		// backlog is full rather than failing it, so without a bound an
+		// overloaded worker turns a refusable condition into a parked peer.
+		relay := newRelay(tun, 0, 50*time.Millisecond)
+		store := NewSessionStore(relay.Stream)
+		DeferCleanup(store.CloseAll)
+		peer, accepted := backloggedPair(256)
+		store.Accept("peer-1", accepted)
+
+		_, frontend := backloggedPair(1)
+		_, err := tun.Attach(ctx, "w1", frontend)
+		Expect(err).ToNot(HaveOccurred())
+		// One un-accepted open fills the one-deep backlog; the relay's own open
+		// is the one that has to wait.
+		filler, err := frontend.OpenStream(ctx)
+		Expect(err).ToNot(HaveOccurred())
+		DeferCleanup(func() { _ = filler.Close() })
+
+		stream, err := peer.OpenStream(ctx)
+		Expect(err).ToNot(HaveOccurred())
+		DeferCleanup(func() { _ = stream.Close() })
+		Expect(WriteRelayRequest(stream, "w1")).To(Succeed())
+
+		replies := make(chan error, 1)
+		go func() {
+			defer GinkgoRecover()
+			replies <- ReadRelayReply(stream)
+		}()
+		var reply error
+		Eventually(replies, "10s").Should(Receive(&reply))
+		Expect(reply).To(MatchError(ErrRelayUnavailable))
+		// The tunnel IS held here, so this must not read as a routing fact.
+		Expect(reply).ToNot(MatchError(ErrNotOwner))
+
+		ends := make(chan error, 1)
+		go func() {
+			defer GinkgoRecover()
+			_, err := stream.Read(make([]byte, 1))
+			ends <- err
+		}()
+		Eventually(ends, "10s").Should(Receive(HaveOccurred()))
+	})
+
+	It("closes the worker's stream when it cannot tell the peer the stream was accepted", func() {
+		// The reply is the last thing that can fail after a worker stream has
+		// been opened. A relay that gave up without closing it would leak one
+		// stream on the worker per failed acceptance, and the worker cannot
+		// tell those from live ones.
+		relay := newRelay(tun, 0, 0)
+		worker, frontend := backloggedPair(256)
+		_, err := tun.Attach(ctx, "w1", frontend)
+		Expect(err).ToNot(HaveOccurred())
+
+		workerSide := make(chan net.Conn, 1)
+		go func() {
+			defer GinkgoRecover()
+			stream, err := worker.AcceptStream()
+			if err != nil {
+				return
+			}
+			workerSide <- stream
+		}()
+
+		peerStream := newUnwritableStream("w1")
+		done := make(chan struct{})
+		go func() {
+			defer GinkgoRecover()
+			defer close(done)
+			relay.Stream("peer-1", peerStream)
+		}()
+		Eventually(done, "10s").Should(BeClosed())
+
+		var served net.Conn
+		Eventually(workerSide, "10s").Should(Receive(&served))
+		ended := make(chan error, 1)
+		go func() {
+			defer GinkgoRecover()
+			_, err := served.Read(make([]byte, 1))
+			ended <- err
+		}()
+		Eventually(ended, "10s").Should(Receive(HaveOccurred()),
+			"the worker-side stream outlived the relay that opened it")
+	})
+})

core/services/cluster/relay_test.go

@@ -288,4 +288,39 @@ var _ = Describe("The relay wire framing", func() {
 	It("refuses to write an empty node id, rather than spending a round trip on it", func() {
 		Expect(cluster.WriteRelayRequest(&bytes.Buffer{}, "")).To(HaveOccurred())
 	})
+
+	// Acceptance is not the whole surface. A refusal read by the wrong hop's
+	// reader must not come back as one of that hop's own sentinels: "the
+	// owning replica does not hold this worker" arriving as "the worker does
+	// not serve that tag" would send a retry to the wrong end of the path, and
+	// it would look like a perfectly ordinary answer on the way.
+	DescribeTable("does not read a relay refusal as one of the worker tunnel's",
+		func(reason error) {
+			frame := &bytes.Buffer{}
+			Expect(cluster.WriteRelayRefusal(frame, reason)).To(Succeed())
+			err := cluster.ReadStreamReply(frame)
+			Expect(err).To(HaveOccurred())
+			Expect(err).ToNot(MatchError(cluster.ErrStreamTagUnknown))
+			Expect(err).ToNot(MatchError(cluster.ErrStreamTargetUnavailable))
+			Expect(err).ToNot(MatchError(cluster.ErrStreamRequestInvalid))
+		},
+		Entry("not the owner", cluster.ErrNotOwner),
+		Entry("the tunnel will not carry a stream", cluster.ErrRelayUnavailable),
+		Entry("a malformed relay request", cluster.ErrRelayRequestInvalid),
+	)
+
+	DescribeTable("does not read a worker tunnel refusal as one of the relay's",
+		func(reason error) {
+			frame := &bytes.Buffer{}
+			Expect(cluster.WriteStreamRefusal(frame, reason)).To(Succeed())
+			err := cluster.ReadRelayReply(frame)
+			Expect(err).To(HaveOccurred())
+			Expect(err).ToNot(MatchError(cluster.ErrNotOwner))
+			Expect(err).ToNot(MatchError(cluster.ErrRelayUnavailable))
+			Expect(err).ToNot(MatchError(cluster.ErrRelayRequestInvalid))
+		},
+		Entry("an unknown stream tag", cluster.ErrStreamTagUnknown),
+		Entry("a local service that will not answer", cluster.ErrStreamTargetUnavailable),
+		Entry("a malformed stream request", cluster.ErrStreamRequestInvalid),
+	)
 })

core/services/cluster/splice.go

@@ -80,27 +80,27 @@ func copyStream(dst io.Writer, src io.Reader) error {
 // *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 isMuxFailure).
+// session's shutdown (see muxVerdict).
 //
 // A socket-level abort (ECONNRESET, EPIPE) is deliberately absent too, which
 // makes the same underlying event, a peer aborting mid-stream, reach the caller
 // as nil over a raw socket where it would be an error. That asymmetry is
-// narrower than it was, since a yamux abort is now reported (see isMuxFailure),
+// narrower than it was, since a yamux abort is now reported (see muxVerdict),
 // and what remains of it is that a raw socket cannot say who aborted.
 //
-// The mux checks run first, and that ordering is load-bearing: a dying yamux
-// session usually hands every live stream its own cause wrapped up
-// (go-yamux/v5@v5.1.0/session.go:328-337), and that cause is routinely a
+// The mux verdict is consulted first, and that ordering is load-bearing: a
+// dying yamux session usually hands every live stream its own cause wrapped up
+// (go-yamux/v5@v5.1.0 session.go, Session.close), 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
 	}
-	if isMuxFailure(err) {
-		return err
-	}
-	if isMuxLocalTeardown(err) {
+	if recognised, report := muxVerdict(err); recognised {
+		if report {
+			return err
+		}
 		return nil
 	}
 	if errors.Is(err, net.ErrClosed) ||
@@ -114,14 +114,24 @@ func normalizeStreamErr(err error) error {
 // declared with it because the constant itself is unexported.
 var normalGoAwayCode = yamux.ErrRemoteGoAway.ErrorCode
 
-// isMuxFailure reports whether err is yamux saying the conversation was CUT,
-// as opposed to ended by this side.
+// muxVerdict classifies a yamux ending. recognised says the error came from the
+// multiplexer at all; report says the ending was INFLICTED on this stream
+// rather than asked for by this side.
 //
-// The distinction is what keeps Splice quiet about the teardown it provokes
-// itself while still reporting a request that died: a keepalive timeout, a
-// broken connection, a peer that reset the stream or a peer that went away
-// under a relayed request has to reach the caller, or a failed inference looks
-// like a finished one.
+// It is ONE function, and that is the point rather than a matter of taste. The
+// policy below turns on a single bit, Remote, and an earlier shape read that
+// bit in two predicates with a report-by-default fallthrough behind them.
+// Reverting either read left the whole suite green, because the error reached
+// the same answer down the other path: the classifier could not be
+// mutation-tested in pieces, which in code whose correctness argument IS its
+// mutation evidence is worse than the duplication it bought. Here each type is
+// decided once, so falsifying either read reddens a spec.
+//
+// The distinction it draws is what keeps Splice quiet about the teardown it
+// provokes itself while still reporting a request that died: a keepalive
+// timeout, a broken connection, a peer that reset the stream or a peer that
+// went away under a relayed request has to reach the caller, or a failed
+// inference looks like a finished one.
 //
 // TWO OF THESE ARE THE POLICY PHASE 1 LEFT OPEN, and this is where they are
 // settled, by the relay in core/services/cluster/relay.go, which is Splice's
@@ -154,6 +164,14 @@ var normalGoAwayCode = yamux.ErrRemoteGoAway.ErrorCode
 // (const.go:96) and is exactly what this process closing its own session
 // produces (session.go:284).
 //
+// The rule is "a remote reset is reported" and not "every remote reset is
+// reported". yamux only builds a *StreamError when the RST rides a
+// typeWindowUpdate frame (stream.go:436); an RST on any other frame type
+// yields the BARE ErrStreamReset sentinel, which is claimed below as this
+// side's own teardown and silenced. Every reset go-yamux itself sends uses
+// typeWindowUpdate, so the gap is unreachable between two LocalAI processes and
+// only a foreign multiplexer implementation could reach it.
+//
 // What this does NOT do is make the far side see a failure. Splice ends both
 // streams with Close, which is a FIN, and a reset after that is a no-op because
 // Close has already moved the stream to streamFinished (stream.go:266-272,
@@ -162,58 +180,44 @@ var normalGoAwayCode = yamux.ErrRemoteGoAway.ErrorCode
 // and HTTP, both of which detect a body that ended without its trailers or its
 // final chunk. Reporting is what the caller needs and this is where it comes
 // from.
-func isMuxFailure(err error) bool {
+func muxVerdict(err error) (recognised, report bool) {
 	// A go-away ends the whole session. Only a normal-code go-away this side
 	// sent is a normal ending.
 	var goAway *yamux.GoAwayError
 	if errors.As(err, &goAway) {
-		return goAway.Remote || goAway.ErrorCode != normalGoAwayCode
+		return true, goAway.Remote || goAway.ErrorCode != normalGoAwayCode
 	}
 	// A stream error is scoped to one stream. Only a reset this side asked for
 	// is a normal ending.
 	var streamErr *yamux.StreamError
 	if errors.As(err, &streamErr) {
-		return streamErr.Remote
+		return true, streamErr.Remote
 	}
-	// Session.close gives every stream it kills ErrStreamReset wrapped around
-	// 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.
+	// Sentinels by identity, never errors.Is, and before the wrapped check
+	// below: these are the endings Splice provokes itself. Closing a stream
+	// whose session has already shut down normally returns ErrSessionShutdown
+	// from the FIN write, which the go-away branch above has already claimed;
+	// a copy parked on a stream that gets closed comes back with
+	// ErrStreamClosed from a Write (stream.go:157-159) or the bare
+	// ErrStreamReset from a Read, which is what CloseRead installs
+	// (stream.go:348-349).
+	if err == yamux.ErrStreamClosed || err == yamux.ErrStreamReset {
+		return true, false
+	}
+	// The same sentinel WRAPPED means something else entirely: Session.close
+	// gives every stream it kills ErrStreamReset wrapped around the cause, so
+	// this is the session dying under a live stream. Identity above is what
+	// separates the two; 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:305-308, 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
-}
-
-// isMuxLocalTeardown reports whether err is yamux ending one stream at this
-// side's request, which Splice mostly provokes itself: closing a stream whose
-// session has already shut down normally returns ErrSessionShutdown from the
-// FIN write, and a copy parked on a stream that gets closed comes back with
-// ErrStreamClosed from a Write (stream.go:157-159) or the bare ErrStreamReset
-// from a Read, which is what CloseRead installs (stream.go:348-349). A reset
-// does not only mean that, though; the same sentinel heads the error a dying
-// session hands its streams, and the remote forms belong to isMuxFailure, which
-// is why that runs first. None of yamux's error types match net.ErrClosed, so
-// all of this has to be recognised by shape.
-func isMuxLocalTeardown(err error) bool {
-	// Sentinels by identity, never errors.Is: the wrapped forms belong to a
-	// dead session and are reported instead. ErrSessionShutdown is absent on
-	// purpose rather than by oversight, being a *GoAwayError carrying the
-	// normal code with Remote false, which the last check below covers.
-	if err == yamux.ErrStreamClosed || err == yamux.ErrStreamReset {
-		return true
+	// as yamux at all, and is why it is left unrecognised here rather than
+	// guessed at: normalizeStreamErr no longer forgives a bare io.EOF, because
+	// for a peer that vanished the raw cause is precisely io.EOF.
+	if errors.Is(err, yamux.ErrStreamReset) {
+		return true, true
 	}
-	// Remote is re-checked rather than assumed from the ordering, so that this
-	// predicate is true to its own name if it is ever called from anywhere
-	// else.
-	var streamErr *yamux.StreamError
-	if errors.As(err, &streamErr) {
-		return !streamErr.Remote
-	}
-	var goAway *yamux.GoAwayError
-	return errors.As(err, &goAway) && !goAway.Remote && goAway.ErrorCode == normalGoAwayCode
+	return false, false
 }