diff --git a/core/services/cluster/peerlink.go b/core/services/cluster/peerlink.go index 047fcbc54..5f5e31864 100644 --- a/core/services/cluster/peerlink.go +++ b/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() diff --git a/core/services/cluster/relay.go b/core/services/cluster/relay.go index 99164c6fb..9fa03807d 100644 --- a/core/services/cluster/relay.go +++ b/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 diff --git a/core/services/cluster/relay_internal_test.go b/core/services/cluster/relay_internal_test.go index 855ac5865..e23f6c55c 100644 --- a/core/services/cluster/relay_internal_test.go +++ b/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") + }) +}) diff --git a/core/services/cluster/relay_test.go b/core/services/cluster/relay_test.go index 389700b7d..e0c7bd049 100644 --- a/core/services/cluster/relay_test.go +++ b/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), + ) }) diff --git a/core/services/cluster/splice.go b/core/services/cluster/splice.go index 691586c25..32fe4358f 100644 --- a/core/services/cluster/splice.go +++ b/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 }