From c19ed5ab3250faa3c29c2bf0941448312d4f35f7 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Wed, 2 Sep 2026 04:48:25 +0000 Subject: [PATCH] fix(cluster): stop a late request frame reading as the worker's verdict Making a worker's refusal reaping evidence created a defect one layer along, at the producer. The worker refused a ReadStreamRequest failure with ErrStreamRequestInvalid and its own comment said "Includes the deadline above expiring", which was harmless while every refusal reached the frontend as "no route" and became a reap the moment one of them did not. So a request frame that had merely not ARRIVED yet was reported as a non-transient verdict about a backend. It is reachable on the relay path, which carries most production traffic: the worker's header timer starts when the OWNING replica opens the stream, while the frame is written by the DIALLING replica only after the relay's acceptance travels back to it, so a whole peer-link round trip runs inside that window, on a link this design deliberately loads with multi-gigabyte artifacts beside token streams. For a long-deadline caller the endpoint is ConnectionEvictingClient, which stops the model across the fleet. It also falsified the "neither clears on its own" argument that licensed the reap. There is now a fourth refusal, ErrStreamNotServed, for what a worker could not serve for a reason of its OWN. It is deliberately outside IsWorkerAnswer, so it reaches a consumer under the no-route umbrella and reaps nothing, which is the same treatment an unrecognised code already gets. Four producers move onto it: a request frame that timed out (a malformed one stays a verdict, because that is a frontend bug no retry fixes), both SetReadDeadline failures, which are facts about the stream and not about a target nothing has dialled yet, and WriteStreamRefusal's default for a reason nobody classified. classifyServiceFailure keeps ErrStreamTargetUnavailable as its default on purpose: inverting it would make errno enumeration the single point of failure for the reap, and a miss there is a row nothing can ever delete. What it gains is a deny-list of two causes that are provably this worker's own clock or its own context. Also: - The read-site caller-deadline guard in the handshake was unpinned: the existing seam spends the budget before the handshake starts, so only the write could ever fail. A spec whose deadline falls between the request and the reply pins it, and each guard now reddens on its own. - The documented worker-first failure line omitted the JSON error envelope the old frontend returns, so an operator grepping it found nothing. - The peer-link disclosure names the aimable per-session receive window in all four places, and LastDialErrorOf records why a third consumer must go through IsWorkerAnswer rather than roll its own list. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/http/endpoints/cluster/peer.go | 16 ++- core/services/cluster/dialer.go | 16 ++- core/services/cluster/dialer_test.go | 56 +++++++++ core/services/cluster/tunnelproto.go | 60 +++++++-- core/services/cluster/tunnelproto_test.go | 57 +++++++-- core/services/nodes/interfaces.go | 18 ++- core/services/nodes/reconciler_prober_test.go | 36 +++++- core/services/nodes/registry.go | 7 +- core/services/worker/tunnel.go | 73 +++++++++-- core/services/worker/tunnel_test.go | 116 ++++++++++++++++++ docs/content/features/distributed-mode.md | 10 +- pkg/grpc/backend.go | 14 +++ 12 files changed, 432 insertions(+), 47 deletions(-) diff --git a/core/http/endpoints/cluster/peer.go b/core/http/endpoints/cluster/peer.go index ca8ef06fd..9dcaea8b3 100644 --- a/core/http/endpoints/cluster/peer.go +++ b/core/http/endpoints/cluster/peer.go @@ -47,12 +47,16 @@ func PeerHandler(token string, onSession func(peerID string, sess *yamux.Session // token (every worker holds it, and it is the same token that // authenticates registration): it can relay to every worker tunnel this // replica owns, reaching every backend gRPC process and every worker's - // file-transfer server; and by declaring a legitimate replica's id it - // can make SessionStore.Accept evict that replica's inbound link, at - // will. Neither is a new capability in KIND - before workers stopped - // listening, a holder of that token could already dial any worker's - // advertised ports directly - but the token is now the only thing - // between an attacker and the whole fleet's tunnels. + // file-transfer server; by declaring a legitimate replica's id it can + // make SessionStore.Accept evict that replica's inbound link, at will; + // and it can aim the per-session receive window, which PeerLinkConfig + // sizes at roughly 31 GiB of unread data per session, at one replica's + // memory. The first two are not new capabilities in KIND - before + // workers stopped listening, a holder of that token could already dial + // any worker's advertised ports directly - but the token is now the + // only thing between an attacker and the whole fleet's tunnels, and the + // third is a figure written down as sizing guidance that is also a + // budget on a route this open. // // It is deferred rather than patched, because the cheap patch does not // work: checking ?id= against the instances table stops an invented id diff --git a/core/services/cluster/dialer.go b/core/services/cluster/dialer.go index 3460dee3a..86c6a8bf0 100644 --- a/core/services/cluster/dialer.go +++ b/core/services/cluster/dialer.go @@ -119,10 +119,18 @@ func isAbsenceClaim(err error) bool { // the worker actually wrote. Everything else it returns is a failure to read // one, which is the tunnel breaking rather than the worker speaking. // -// A reply carrying a code this frontend does not recognise is deliberately NOT -// counted here, even though a worker did send it. Classifying it as an answer -// would let a newer worker's vocabulary be read by an older frontend as -// evidence about a backend, and the consequence of guessing wrong in that +// ErrStreamNotServed is deliberately NOT here, even though it is the fourth +// refusal and a worker plainly sent it. It is the code a worker uses to say it +// learned nothing: a request frame that never arrived in time, a stream whose +// deadline could not be armed, anything it could not classify. Those clear on +// their own, so they must reach a consumer as "no route" and cost a retry, not +// a row. Keeping the fourth code out of this predicate is what makes it +// possible for the worker to answer honestly at all. +// +// A reply carrying a code this frontend does not recognise is NOT counted here +// either, for the same reason at the next version boundary. Classifying it as +// an answer would let a newer worker's vocabulary be read by an older frontend +// as evidence about a backend, and the consequence of guessing wrong in that // direction is a reaped replica; guessing wrong the other way costs a retry. // // It is EXPORTED because it is half of a contract, not an implementation diff --git a/core/services/cluster/dialer_test.go b/core/services/cluster/dialer_test.go index e9ce9b243..8edd28e52 100644 --- a/core/services/cluster/dialer_test.go +++ b/core/services/cluster/dialer_test.go @@ -325,6 +325,62 @@ var _ = Describe("The worker dialer", func() { expectNoRoute(out.err) }) + It("blames the caller's spent budget when it runs out BETWEEN the request and the reply", func() { + // R2: the read-site guard, which its sibling above cannot reach. + // + // deadlinePassed makes the budget already spent when handshake + // starts, so the WRITE is what fails and only the write-site guard + // fires. The guard that matters in production is the other one: a + // caller with a real budget writes its request successfully, the + // worker takes longer than the remainder to answer, and the READ + // ends on the deadline handshakeDeadline armed from that same + // budget. Deleting the read-site guard left the whole cluster suite + // green, which is exactly the "pinned at one of three sites" gap + // this branch has now closed twice. + // + // Deterministic without a sleep: the worker reads the request and + // then never answers, so nothing but the caller's own deadline can + // end the exchange, and the spec waits on the request having been + // SEEN before waiting on the dial. Seeing it is also what proves + // the write succeeded and therefore that this is the read site. + frontend, worker := workerTunnel() + _, err := mine.Attach(ctx, "w1", frontend) + Expect(err).ToNot(HaveOccurred()) + + seen := make(chan servedRequest, 1) + go func() { + defer GinkgoRecover() + stream, err := worker.AcceptStream() + if err != nil { + seen <- servedRequest{err: err} + return + } + tag, target, err := cluster.ReadStreamRequest(stream) + seen <- servedRequest{tag: tag, target: target, stream: stream, err: err} + // Deliberately no reply. The caller's deadline is the only + // thing left that can end this handshake. + }() + + budgeted, cancel := context.WithTimeout(ctx, 750*time.Millisecond) + defer cancel() + d := cluster.NewWorkerDialer(mine, nil) + result := dialAsync(d, budgeted, "w1", cluster.StreamTagGRPC, "127.0.0.1:41000") + + var req servedRequest + Eventually(seen, "10s").Should(Receive(&req)) + Expect(req.err).ToNot(HaveOccurred(), + "the request frame must have been written and read, or this spec is testing the write site") + + var out dialResult + Eventually(result, "10s").Should(Receive(&out)) + Expect(out.err).To(MatchError(context.DeadlineExceeded), + "the caller ran out waiting for a reply; the worker never gave a verdict") + Expect(out.err.Error()).To(ContainSubstring("the caller's own budget ran out")) + Expect(out.err.Error()).To(ContainSubstring(`opening "grpc"`), + "this is the READ site; the write site says \"asking for\"") + expectNoRoute(out.err) + }) + It("reports a broken tunnel held here as itself, not as a routing fact", func() { // ErrNotOwner tells a caller to look for the worker elsewhere. For // a tunnel held right here that sends it back to this replica, and diff --git a/core/services/cluster/tunnelproto.go b/core/services/cluster/tunnelproto.go index c6433459f..33daac5db 100644 --- a/core/services/cluster/tunnelproto.go +++ b/core/services/cluster/tunnelproto.go @@ -59,19 +59,24 @@ const ( replyCodeUnknownTag = "unknown-tag" replyCodeUnavailable = "unavailable" replyCodeBadRequest = "bad-request" + replyCodeNotServed = "not-served" replyPrefixRefused = "err " streamRequestSeparator = " " ) -// The three refusals a worker can send, kept apart on purpose. +// The four refusals a worker can send, kept apart on purpose. // // This is the phase's standing rule in its wire form. An unknown tag is a fact // about what this worker SERVES and will not change until the worker is -// upgraded; an unavailable target is an infrastructure failure that may well -// succeed on the next attempt; a bad request is this frontend's own bug. A -// caller retries the second, gives up on the first, and reports the third. -// Collapsing them into one error would make a frontend retry a stream that can -// never work, or abandon a backend that was merely restarting. +// upgraded; an unavailable target is the worker's own dial to the named +// process failing, which is what a backend that died looks like from inside +// the worker; a bad request is this frontend's own bug. A caller gives up on +// the first, acts on the second, and reports the third. Collapsing them into +// one error would make a frontend retry a stream that can never work, or +// abandon a backend that was merely restarting. +// +// The fourth is the one that says NOTHING, and it exists because the first +// three all say something a consumer now acts on. See ErrStreamNotServed. // // None of them wraps a node-absence error, and none must ever be built over // one: a refusal is proof the worker is CONNECTED and answered. @@ -79,6 +84,33 @@ var ( ErrStreamTagUnknown = errors.New("cluster: the worker does not serve that stream tag") ErrStreamTargetUnavailable = errors.New("cluster: the worker could not reach the local service for that stream") ErrStreamRequestInvalid = errors.New("cluster: the worker rejected the stream request as malformed") + + // ErrStreamNotServed reports that the worker could not serve the stream for + // a reason of ITS OWN, which is not a statement about the backend the + // stream named. + // + // It is the refusal a worker sends when it has learned nothing. The other + // three are evidence a frontend ACTS on: IsWorkerAnswer exempts them from + // the no-route umbrella, and nodes.unroutable then lets a reap guard delete + // the row. This one is deliberately OUTSIDE that predicate, so it reaches a + // consumer as ErrNoRoute and nothing is reaped. + // + // It exists because of a defect this phase created and then found: the + // worker used to answer a request frame that merely arrived LATE with + // ErrStreamRequestInvalid, which was harmless while the frontend treated + // every refusal as "no route", and became a reap the moment a frontend + // started acting on refusals. A delivery timeout clears as soon as the link + // drains; a malformed frame does not. Merging them was safe only while + // nothing downstream could tell them apart, and something downstream now + // can. Anything the worker cannot classify as one of the other three + // belongs here, because an unclassified failure is by definition not a + // verdict about a backend. + // + // A frontend too old to know this code reads it as an unrecognised reply, + // which ReadStreamReply already returns as a plain error and which + // IsWorkerAnswer already declines to count as an answer. So the safe + // behaviour is what a mixed-version deployment gets for free. + ErrStreamNotServed = errors.New("cluster: the worker could not serve that stream, for a reason that is not about the backend") ) // WriteStreamRequest sends the opening frame naming what the stream is for. @@ -127,12 +159,20 @@ func WriteStreamAccepted(w io.Writer) error { // WriteStreamRefusal reports why a stream will not be served. The caller closes // the stream afterwards; this only says why. // -// An unrecognised reason is sent as bad-request with its text attached rather +// An unrecognised reason is sent as NOT-SERVED with its text attached rather // than being dropped, because a refusal a frontend cannot read is // indistinguishable from a worker that hung up, and those are different // problems. +// +// The default is not-served and not bad-request, and the difference is the +// whole point of the fourth code. The other three are evidence a frontend acts +// on, up to and including deleting a model's row; an error that reached here +// without carrying one of them is by construction an error nobody classified, +// and an unclassified failure must never become a verdict by default. This +// default used to be bad-request, which was harmless while no consumer +// distinguished the codes and became a reap-by-omission when one did. func WriteStreamRefusal(w io.Writer, reason error) error { - code := replyCodeBadRequest + code := replyCodeNotServed switch { case errors.Is(reason, ErrStreamTagUnknown): code = replyCodeUnknownTag @@ -140,6 +180,8 @@ func WriteStreamRefusal(w io.Writer, reason error) error { code = replyCodeUnavailable case errors.Is(reason, ErrStreamRequestInvalid): code = replyCodeBadRequest + case errors.Is(reason, ErrStreamNotServed): + code = replyCodeNotServed } text := "" @@ -186,6 +228,8 @@ func ReadStreamReply(r io.Reader) error { return fmt.Errorf("%w: %s", ErrStreamTargetUnavailable, text) case replyCodeBadRequest: return fmt.Errorf("%w: %s", ErrStreamRequestInvalid, text) + case replyCodeNotServed: + return fmt.Errorf("%w: %s", ErrStreamNotServed, text) default: // A code from a newer worker. Reported as an error carrying the code // rather than mapped onto the nearest known one, so a frontend does not diff --git a/core/services/cluster/tunnelproto_test.go b/core/services/cluster/tunnelproto_test.go index 2664d9d50..2c7e8d5d0 100644 --- a/core/services/cluster/tunnelproto_test.go +++ b/core/services/cluster/tunnelproto_test.go @@ -3,6 +3,7 @@ package cluster_test import ( "bytes" "encoding/binary" + "errors" "io" "strings" "unicode/utf8" @@ -100,27 +101,66 @@ var _ = Describe("Worker tunnel stream framing", func() { Expect(cluster.ReadStreamReply(&buf)).To(Succeed()) }) - DescribeTable("keeps the three refusals apart", + DescribeTable("keeps the four refusals apart", func(sent error, others []error) { var buf bytes.Buffer Expect(cluster.WriteStreamRefusal(&buf, sent)).To(Succeed()) got := cluster.ReadStreamReply(&buf) Expect(got).To(MatchError(sent)) - // The whole point. A caller gives up on an unknown tag, retries - // an unavailable target, and reports a bad request as its own - // bug; collapsing any pair makes one of those wrong. + // The whole point. A caller gives up on an unknown tag, acts on + // an unavailable target, reports a bad request as its own bug, + // and learns NOTHING from a not-served; collapsing any pair + // makes one of those wrong, and three of the four pairs end in + // a reaped replica. for _, other := range others { Expect(got).ToNot(MatchError(other)) } }, Entry("unknown tag", cluster.ErrStreamTagUnknown, - []error{cluster.ErrStreamTargetUnavailable, cluster.ErrStreamRequestInvalid}), + []error{cluster.ErrStreamTargetUnavailable, cluster.ErrStreamRequestInvalid, cluster.ErrStreamNotServed}), Entry("unavailable target", cluster.ErrStreamTargetUnavailable, - []error{cluster.ErrStreamTagUnknown, cluster.ErrStreamRequestInvalid}), + []error{cluster.ErrStreamTagUnknown, cluster.ErrStreamRequestInvalid, cluster.ErrStreamNotServed}), Entry("invalid request", cluster.ErrStreamRequestInvalid, - []error{cluster.ErrStreamTagUnknown, cluster.ErrStreamTargetUnavailable}), + []error{cluster.ErrStreamTagUnknown, cluster.ErrStreamTargetUnavailable, cluster.ErrStreamNotServed}), + Entry("nothing learned", cluster.ErrStreamNotServed, + []error{cluster.ErrStreamTagUnknown, cluster.ErrStreamTargetUnavailable, cluster.ErrStreamRequestInvalid}), ) + It("sends a reason it cannot classify as not-served, never as a verdict", func() { + // The default, and it is a safety default rather than a formality. + // Three of the four codes are evidence a frontend now ACTS on, up + // to deleting a model's row; an error that reached WriteStreamRefusal + // without carrying a sentinel is by construction one nobody + // classified. This default used to be bad-request, which was + // harmless while no consumer distinguished the codes and became a + // reap-by-omission the moment one did. + var buf bytes.Buffer + Expect(cluster.WriteStreamRefusal(&buf, errors.New("something nobody thought about"))).To(Succeed()) + got := cluster.ReadStreamReply(&buf) + Expect(got).To(MatchError(cluster.ErrStreamNotServed)) + Expect(got).ToNot(MatchError(cluster.ErrStreamRequestInvalid)) + Expect(cluster.IsWorkerAnswer(got)).To(BeFalse(), + "an unclassified failure must never become the worker's verdict about a backend") + Expect(got.Error()).To(ContainSubstring("something nobody thought about")) + }) + + It("keeps not-served OUT of the answers a frontend acts on", func() { + // The predicate is the seam between what the worker says and what + // the frontend does with it. The other three are exempted from the + // no-route umbrella so a crashed backend can be reaped; this one + // must not be, or every transient worker-side failure reaps. + var buf bytes.Buffer + Expect(cluster.WriteStreamRefusal(&buf, cluster.ErrStreamNotServed)).To(Succeed()) + Expect(cluster.IsWorkerAnswer(cluster.ReadStreamReply(&buf))).To(BeFalse()) + + for _, verdict := range []error{cluster.ErrStreamTagUnknown, cluster.ErrStreamTargetUnavailable, cluster.ErrStreamRequestInvalid} { + buf.Reset() + Expect(cluster.WriteStreamRefusal(&buf, verdict)).To(Succeed()) + Expect(cluster.IsWorkerAnswer(cluster.ReadStreamReply(&buf))).To(BeTrue(), + "a verdict that stopped being an answer makes a dead backend unreapable") + } + }) + It("carries the reason text to the far side", func() { var buf bytes.Buffer Expect(cluster.WriteStreamRefusal(&buf, wrapReason(cluster.ErrStreamTagUnknown, "no-such-tag"))).To(Succeed()) @@ -139,6 +179,8 @@ var _ = Describe("Worker tunnel stream framing", func() { Expect(got).ToNot(MatchError(cluster.ErrStreamTagUnknown)) Expect(got).ToNot(MatchError(cluster.ErrStreamTargetUnavailable)) Expect(got).ToNot(MatchError(cluster.ErrStreamRequestInvalid)) + Expect(got).ToNot(MatchError(cluster.ErrStreamNotServed)) + Expect(cluster.IsWorkerAnswer(got)).To(BeFalse()) }) It("reports a failure to READ the reply as itself, never as a refusal", func() { @@ -150,6 +192,7 @@ var _ = Describe("Worker tunnel stream framing", func() { Expect(got).ToNot(MatchError(cluster.ErrStreamTagUnknown)) Expect(got).ToNot(MatchError(cluster.ErrStreamTargetUnavailable)) Expect(got).ToNot(MatchError(cluster.ErrStreamRequestInvalid)) + Expect(got).ToNot(MatchError(cluster.ErrStreamNotServed)) }) It("truncates an over-long reason on a rune boundary, keeping it decodable", func() { diff --git a/core/services/nodes/interfaces.go b/core/services/nodes/interfaces.go index 9a627cdf7..c481113c3 100644 --- a/core/services/nodes/interfaces.go +++ b/core/services/nodes/interfaces.go @@ -233,9 +233,21 @@ var ErrNoWorkerDialer = fmt.Errorf("%w: no worker tunnel dialer is configured", // EVERY replica for as long as it exists, and reaping it converges: the model // is reloaded somewhere that works and re-registers a usable address. The // condition the phase refuses to reap on is a TRANSIENT one, and none of these -// is transient. A reply code this frontend does not recognise is deliberately -// not in the set (see cluster.IsWorkerAnswer), so a newer worker's vocabulary -// reaches an older frontend as "no route" and costs a retry rather than a row. +// is transient. +// +// That last sentence is a claim about the WORKER, not about this file, and it +// held only after the worker stopped answering a request frame that merely +// arrived late with ErrStreamRequestInvalid. It did, and the frontend's half of +// that contract is that a transient condition arrives as the fourth code: +// cluster.ErrStreamNotServed is not in cluster.IsWorkerAnswer, so it reaches +// here under the no-route umbrella and reaps nothing. If a worker ever starts +// sending one of the three for something that clears on its own, this comment +// becomes false and a live model gets evicted; the guard against that is at the +// worker, in Tunnel.accept and classifyServiceFailure, and it is stated there. +// +// A reply code this frontend does not recognise is deliberately not in the set +// either (see cluster.IsWorkerAnswer), so a newer worker's vocabulary reaches +// an older frontend as "no route" and costs a retry rather than a row. func unroutable(client grpc.Backend) error { // LastDialErrorOf and not a type assertion: the assertion could not see // past a decorator, and SmartRouter hands every routed client out wrapped. diff --git a/core/services/nodes/reconciler_prober_test.go b/core/services/nodes/reconciler_prober_test.go index a45a13093..b8cb80750 100644 --- a/core/services/nodes/reconciler_prober_test.go +++ b/core/services/nodes/reconciler_prober_test.go @@ -26,9 +26,16 @@ import ( // refusal is written with the worker's own writer and read back with the // frontend's own reader, so a reason the protocol cannot carry, or a code // mapping that stopped round-tripping, reddens these specs instead of leaving -// them asserting against a value production never produces. The final wrap is -// the shape WorkerDialer.handshake returns for a worker answer, umbrella and -// all: there is none, which is the property under test. +// them asserting against a value production never produces. +// +// The wrap is chosen by cluster.IsWorkerAnswer, which is what +// WorkerDialer.handshake does, so the spec exercises the real branch rather +// than a transcription of it. That is not circular: the helper decides the +// SHAPE of the error and the table below states the OUTCOME independently, so +// moving a sentinel into or out of the predicate reddens the table. In +// particular, adding ErrStreamNotServed to the predicate would strip its +// umbrella here and turn its ProbeUnknown entry red, which is the property that +// keeps a transient worker-side failure from reaping a live model. func refusalFromWorker(reason error) error { GinkgoHelper() var frame bytes.Buffer @@ -36,7 +43,10 @@ func refusalFromWorker(reason error) error { readBack := cluster.ReadStreamReply(&frame) Expect(readBack).To(MatchError(reason), "the refusal must survive its own round trip") Expect(readBack).ToNot(MatchError(cluster.ErrNoRoute)) - return fmt.Errorf("opening %q on node %q: %w", "grpc", "node-1", readBack) + if cluster.IsWorkerAnswer(readBack) { + return fmt.Errorf("opening %q on node %q: %w", "grpc", "node-1", readBack) + } + return fmt.Errorf("reaching node %q: %w: opening %q: %w", "node-1", cluster.ErrNoRoute, "grpc", readBack) } // proberFactory hands the prober one client, and records what it was asked for. @@ -122,6 +132,24 @@ var _ = Describe("the reconciler's gRPC model prober", func() { Entry("the worker rejected the stored address", cluster.ErrStreamRequestInvalid), ) + It("answers ProbeUnknown when the worker refused but said it learned nothing", func() { + // The fourth refusal, and the boundary that keeps the three above safe + // to act on. A worker answers with it for its OWN transient conditions: + // a request frame that never arrived in time, a stream whose deadline + // would not arm, a local dial that ended on the session going away. + // Those clear on a reconnect, so acting on them would convert + // peer-link congestion into a reaped row, and on the inference path + // into a model stopped across the fleet. + // + // This is not hypothetical: the header-timeout case USED to arrive as + // ErrStreamRequestInvalid, which the table above reaps on. + Expect(probe(&proberFactory{client: &fakeBackendClient{ + healthy: false, + err: status.Error(codes.Unavailable, "connection error: transport"), + dialErr: refusalFromWorker(cluster.ErrStreamNotServed), + }})).To(Equal(ProbeUnknown)) + }) + It("answers ProbeUnknown for a refusal code this frontend does not recognise", func() { // The other direction, and the boundary of the exemption above. A // newer worker's vocabulary must not be read as evidence about a diff --git a/core/services/nodes/registry.go b/core/services/nodes/registry.go index 00571e6ab..036b4b558 100644 --- a/core/services/nodes/registry.go +++ b/core/services/nodes/registry.go @@ -51,9 +51,10 @@ type BackendNode struct { // still true. A leaked registration token no longer lets its holder BE a // worker; it still lets its holder REACH every worker, because // GET /api/cluster/peer authenticates with the shared cluster token and - // takes its ?id= on trust (see core/http/endpoints/cluster/peer.go). Per - // replica-to-replica credentials are a named phase-3 item, not something - // this column already delivers. + // takes its ?id= on trust (see core/http/endpoints/cluster/peer.go), which + // also puts the ~31 GiB per-session peer receive window inside reach of + // anything holding it. Per replica-to-replica credentials are a named + // phase-3 item, not something this column already delivers. // // Empty means no tunnel credential has been minted for this node yet, which // is what a node registered by an older LocalAI looks like. Such a node diff --git a/core/services/worker/tunnel.go b/core/services/worker/tunnel.go index 0ff28ea85..cfd33a770 100644 --- a/core/services/worker/tunnel.go +++ b/core/services/worker/tunnel.go @@ -9,6 +9,7 @@ import ( "net" "net/http" "net/url" + "os" "strconv" "strings" "sync" @@ -335,17 +336,32 @@ func (t *Tunnel) accept(ctx context.Context, stream net.Conn) (net.Conn, bool) { // Deliberately time.Now and not t.now: this is an I/O deadline, and the // clock seam exists only to measure how long a session lasted. if err := stream.SetReadDeadline(time.Now().Add(t.headerTimeout)); err != nil { - // Nothing is readable on a stream whose deadline cannot be set, so this - // is reported as an infrastructure failure rather than pushed past. - t.refuse(stream, fmt.Errorf("%w: arming the request deadline: %v", cluster.ErrStreamTargetUnavailable, err)) + // NotServed and not TargetUnavailable: this is a fact about the STREAM, + // which would not take a deadline, and no local service has been named + // yet, let alone dialled. Reporting it as an unreachable target would + // tell the frontend a backend it has not asked about is gone. + t.refuse(stream, fmt.Errorf("%w: arming the request deadline: %v", cluster.ErrStreamNotServed, err)) return nil, false } tag, target, err := cluster.ReadStreamRequest(stream) if err != nil { - // Includes the deadline above expiring. Both are "this stream never - // told me what it wanted", which is the frontend's problem to fix, not - // something a retry against this worker resolves. + // The two causes are SEPARATED here, and merging them was a real + // defect. A malformed frame is the frontend's own bug and does not + // clear on its own, so it stays a verdict the frontend acts on. The + // deadline above expiring is a frame that has not ARRIVED yet, which + // clears the moment the link drains; on the relay path the worker's + // timer starts when the OWNING replica opens the stream, while the + // frame is written by the DIALLING replica only after the relay's + // acceptance has travelled back to it, so a whole peer-link round trip + // runs inside this window, on a link this design deliberately loads + // with multi-gigabyte artifacts. Reported as a malformed request it + // became reaping evidence, and for a long-deadline caller that is a + // model evicted across the fleet by nothing but congestion. + if isReadTimeout(err) { + t.refuse(stream, fmt.Errorf("%w: %v", cluster.ErrStreamNotServed, err)) + return nil, false + } t.refuse(stream, fmt.Errorf("%w: %v", cluster.ErrStreamRequestInvalid, err)) return nil, false } @@ -365,7 +381,9 @@ func (t *Tunnel) accept(ctx context.Context, stream net.Conn) (net.Conn, bool) { // its own deadlines, and one left armed here would abort a long inference // stream in the middle. if err := stream.SetReadDeadline(time.Time{}); err != nil { - t.refuse(stream, fmt.Errorf("%w: clearing the request deadline: %v", cluster.ErrStreamTargetUnavailable, err)) + // NotServed for the same reason as arming it: the stream is what + // failed, and this worker has said nothing about the target. + t.refuse(stream, fmt.Errorf("%w: clearing the request deadline: %v", cluster.ErrStreamNotServed, err)) return nil, false } @@ -397,16 +415,51 @@ func (t *Tunnel) accept(ctx context.Context, stream net.Conn) (net.Conn, bool) { // retry something that can never work; reporting the second as the first makes // it give up on a backend that is merely starting. // -// Anything unclassified is infrastructure, because that is what an unadorned -// dial failure is, and it must never become the unknown-tag refusal: a tag this -// worker serves does not stop being served because one dial failed. +// The default is TargetUnavailable and stays that way, which is the deliberate +// half of this function now that the frontend acts on that code. A dial to the +// named process that came back with anything is the closest thing to evidence +// this worker can produce, and the direction of a mis-classification decides +// which mistake is made: defaulting the other way would mean a genuinely dead +// backend whose errno nobody enumerated becomes a permanently unreapable row, +// which is the exact defect this phase spent a round removing. So the +// exemptions below are a DENY-list of causes that are provably not about the +// target, not an allow-list of causes that are. +// +// The two exempted are the caller's context ending and a read/write deadline +// firing. Neither is the target answering: the context here is the SESSION's, +// so it ends when this worker's tunnel is torn down, and a deadline is this +// process's own timer. Both clear on a reconnect. They are exempted rather +// than argued away as unreachable because "unreachable" was the argument that +// made the request-frame merge look safe. +// +// It must never become the unknown-tag refusal either: a tag this worker serves +// does not stop being served because one dial failed. func classifyServiceFailure(err error) error { if errors.Is(err, cluster.ErrStreamRequestInvalid) { return err } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || isReadTimeout(err) { + return fmt.Errorf("%w: %v", cluster.ErrStreamNotServed, err) + } return fmt.Errorf("%w: %v", cluster.ErrStreamTargetUnavailable, err) } +// isReadTimeout reports whether err is an I/O deadline firing rather than a +// peer saying anything. +// +// net.Error's Timeout is asked as well as os.ErrDeadlineExceeded because the +// two are not the same set: a yamux stream returns its own timeout value from +// a Read whose deadline expired, and a net.OpError over a socket returns +// os.ErrDeadlineExceeded. Missing either would put a timeout back on the +// verdict path, which is the defect this predicate exists to keep closed. +func isReadTimeout(err error) bool { + if errors.Is(err, os.ErrDeadlineExceeded) { + return true + } + var netErr net.Error + return errors.As(err, &netErr) && netErr.Timeout() +} + // refuse reports why a stream will not be served and then ENDS it. // // The close is the part that matters and it is not optional. A worker that says diff --git a/core/services/worker/tunnel_test.go b/core/services/worker/tunnel_test.go index 8a59f686f..743cccf72 100644 --- a/core/services/worker/tunnel_test.go +++ b/core/services/worker/tunnel_test.go @@ -12,6 +12,7 @@ import ( "strconv" "strings" "sync/atomic" + "syscall" "time" "github.com/gorilla/websocket" @@ -349,6 +350,121 @@ var _ = Describe("Worker tunnel client", func() { }) Eventually(ended, "10s").Should(Receive(BeNil())) }) + + It("says it learned NOTHING when the request frame never arrived in time", func() { + // The producer side of the phase's worst self-inflicted defect. + // + // This refusal used to be ErrStreamRequestInvalid, merged with a + // genuinely malformed frame on the grounds that both are "this + // stream never told me what it wanted". That was safe only while + // the frontend treated every refusal as "no route". Once + // nodes.unroutable started exempting worker answers so a crashed + // backend could be reaped, this became reaping evidence for a frame + // that had merely not ARRIVED yet. + // + // It is reachable: on the relay path the worker's header timer + // starts when the OWNING replica opens the stream, while the frame + // is written by the DIALLING replica only after the relay + // acceptance travels back, so a peer-link round trip runs inside + // this window on a link that also carries multi-gigabyte artifacts. + // For a long-deadline caller the endpoint is + // ConnectionEvictingClient, which stops the model across the fleet. + frontend = newFakeFrontend(false) + start(func(c *TunnelConfig) { + c.Services[cluster.StreamTagGRPC] = dialLocalTCP + c.headerTimeout = 50 * time.Millisecond + }) + + stream, err := session().OpenStream(ctx) + Expect(err).ToNot(HaveOccurred()) + // Nothing is written, so only the header timer can end this. + reply := awaitErr(func() error { return cluster.ReadStreamReply(stream) }) + var got error + Eventually(reply, "10s").Should(Receive(&got)) + + Expect(got).To(MatchError(cluster.ErrStreamNotServed)) + // The three assertions that make this bite. Each of the other + // sentinels is evidence the frontend acts on, and the predicate is + // the single place the two lists are kept identical. + Expect(got).ToNot(MatchError(cluster.ErrStreamRequestInvalid), + "a frame that arrived late is not a malformed frame, and this one reaps") + Expect(got).ToNot(MatchError(cluster.ErrStreamTargetUnavailable)) + Expect(cluster.IsWorkerAnswer(got)).To(BeFalse(), + "a timeout must reach a reap guard as no-route, never as the worker's verdict") + }) + + It("still calls a MALFORMED request frame malformed, which is a verdict", func() { + // The other direction. Separating the timeout out must not turn the + // verdict off: a frontend that writes a frame this worker cannot + // parse has a bug that no retry fixes, and the refusal has to keep + // saying so. + frontend = newFakeFrontend(false) + start(func(c *TunnelConfig) { + c.Services[cluster.StreamTagGRPC] = dialLocalTCP + c.headerTimeout = time.Minute + }) + + stream, err := session().OpenStream(ctx) + Expect(err).ToNot(HaveOccurred()) + // A frame whose declared length exceeds what the reader will take, + // so the failure is the frame's shape and not the clock. + Expect(binary.Write(stream, binary.BigEndian, uint16(60000))).To(Succeed()) + + reply := awaitErr(func() error { return cluster.ReadStreamReply(stream) }) + var got error + Eventually(reply, "10s").Should(Receive(&got)) + + Expect(got).To(MatchError(cluster.ErrStreamRequestInvalid)) + Expect(got).ToNot(MatchError(cluster.ErrStreamNotServed)) + Expect(cluster.IsWorkerAnswer(got)).To(BeTrue()) + }) + + It("says it learned nothing when the local dial ended on the session going away", func() { + // classifyServiceFailure's deny-list. The default there is + // TargetUnavailable and stays that way, because a mis-classified + // dial failure must fall towards "reapable" rather than towards a + // row nothing can ever delete. What is exempted is the pair of + // causes that are provably not the target answering: this worker's + // own session context ending, and its own I/O deadline firing. + frontend = newFakeFrontend(false) + start(func(c *TunnelConfig) { + c.Services[cluster.StreamTagGRPC] = func(ctx context.Context, _ string) (net.Conn, error) { + return nil, fmt.Errorf("dialing the backend: %w", context.Canceled) + } + }) + + stream, err := session().OpenStream(ctx) + Expect(err).ToNot(HaveOccurred()) + Expect(cluster.WriteStreamRequest(stream, cluster.StreamTagGRPC, "127.0.0.1:41000")).To(Succeed()) + + reply := awaitErr(func() error { return cluster.ReadStreamReply(stream) }) + var got error + Eventually(reply, "10s").Should(Receive(&got)) + Expect(got).To(MatchError(cluster.ErrStreamNotServed)) + Expect(cluster.IsWorkerAnswer(got)).To(BeFalse()) + }) + + It("still reports a refused local dial as an unavailable target, which reaps", func() { + // The other direction for the deny-list: the ordinary shape of a + // crashed backend must keep producing the code the reap guards act + // on, or the ghost rows come back. + frontend = newFakeFrontend(false) + start(func(c *TunnelConfig) { + c.Services[cluster.StreamTagGRPC] = func(context.Context, string) (net.Conn, error) { + return nil, fmt.Errorf("dial tcp 127.0.0.1:41000: connect: %w", syscall.ECONNREFUSED) + } + }) + + stream, err := session().OpenStream(ctx) + Expect(err).ToNot(HaveOccurred()) + Expect(cluster.WriteStreamRequest(stream, cluster.StreamTagGRPC, "127.0.0.1:41000")).To(Succeed()) + + reply := awaitErr(func() error { return cluster.ReadStreamReply(stream) }) + var got error + Eventually(reply, "10s").Should(Receive(&got)) + Expect(got).To(MatchError(cluster.ErrStreamTargetUnavailable)) + Expect(cluster.IsWorkerAnswer(got)).To(BeTrue()) + }) }) Describe("surviving a bad stream", func() { diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index 94255735e..636ecde49 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -114,7 +114,7 @@ environment: The peer link is served at `/api/cluster/peer` and authenticates with `LOCALAI_REGISTRATION_TOKEN`, the same shared secret workers register with. Replicas that disagree about it cannot link. A replica that stops heartbeating for 30 seconds is dropped from the table by the others, along with the worker-connection rows it owned. {{% notice note %}} -**The peer link has no per-replica credential yet.** It checks the shared registration token and takes the replica id in `?id=` on trust. Anything already holding that token - every worker holds it - can therefore open a peer link, relay through it to every worker tunnel a replica owns, and by declaring another replica's id displace that replica's inbound link. Treat `LOCALAI_REGISTRATION_TOKEN` as a cluster-wide secret with the blast radius of the whole fleet: give it its own value per deployment, do not reuse it elsewhere, and keep `/api/cluster/peer` on a network only your replicas and workers can reach. Per-replica credentials for this route are planned. +**The peer link has no per-replica credential yet.** It checks the shared registration token and takes the replica id in `?id=` on trust. Anything already holding that token - every worker holds it - can therefore open a peer link, relay through it to every worker tunnel a replica owns, by declaring another replica's id displace that replica's inbound link, and hold sessions open against the per-session receive window, which the peer-link code sizes at roughly 31 GiB of unread data per session and which on this route is also a memory budget an attacker can point at one replica. Treat `LOCALAI_REGISTRATION_TOKEN` as a cluster-wide secret with the blast radius of the whole fleet: give it its own value per deployment, do not reuse it elsewhere, and keep `/api/cluster/peer` on a network only your replicas and workers can reach. Per-replica credentials for this route are planned. {{% /notice %}} ### Worker tunnels @@ -207,7 +207,13 @@ That distinction is the whole point rather than a nicety. A scheduler told that - **Frontends first (correct).** Old workers keep running, keep heartbeating and keep their `node_models` rows: the new frontend reports them as unroutable rather than as gone, so nothing is rescheduled and nothing is reaped. What fails is requests for models on a worker that has not been restarted yet. That is a real degraded window, but it is bounded by how fast you roll the workers, it heals itself as each one comes back, and no state is lost. - **What you will see while it lasts:** requests for models on a not-yet-restarted worker fail with "no route to the worker", while `GET /api/nodes` still shows that node healthy and heartbeating and its models still listed. Restart the worker and it clears. Nothing needs fixing; you are watching the window close. - **Workers first (this fails, do not do it).** An old frontend has no `/api/cluster/connect` route for the worker to dial *and* rejects the new worker's registration outright, because the worker no longer sends an address and the old frontend requires one. A 4xx is a verdict rather than an outage, so the worker reports the reason on the **first** attempt and exits instead of retrying. Every worker you restart is a worker you take out of the fleet until the frontends are upgraded. - - **What you will see if you do it anyway:** each restarted worker exits within a second or two of starting, with `registration failed with status 400: address is required for backend workers: the frontend refused this registration`. The fleet drains one node per restart, and the nodes that are left are the ones you have not touched yet. + - **What you will see if you do it anyway:** each restarted worker exits within a second or two of starting, with + + ``` + registration failed with status 400: {"error":{"code":400,"message":"address is required for backend workers","type":"node_error"}}: the frontend refused this registration + ``` + + The fleet drains one node per restart, and the nodes that are left are the ones you have not touched yet. Grep for `address is required for backend workers` if your log collector reflows the line. A worker that cannot reach its frontend *at the network level* retries with exponential backoff and never gives up, so restarting a worker is all that is needed to close the frontend-first window. A worker whose registration is **rejected** does not retry, which is what makes the wrong order destructive rather than slow. diff --git a/pkg/grpc/backend.go b/pkg/grpc/backend.go index a0fdd2ad8..42b46e39b 100644 --- a/pkg/grpc/backend.go +++ b/pkg/grpc/backend.go @@ -133,6 +133,20 @@ const maxBackendUnwrapDepth = 16 // pkg/model is an *InFlightTrackingClient over a *FileStagingClient over the // real one, so both callers were asking a wrapper that had no answer and // reading nil as "the transport was fine". +// +// WHAT IT ANSWERS IS NOT "was this the transport's fault". It answers "what did +// the dialler last return", and in distributed mode some of those values are a +// WORKER'S OWN REFUSAL, which means the tunnel worked and the worker spoke. +// Telling those apart is cluster.IsWorkerAnswer, and the two production callers +// (nodes.unroutable, model.transportFailure) both go through it. A new caller +// that matches on sentinels of its own would be re-creating the collapse this +// phase spent two rounds removing: the reap guards and the dialler would stop +// agreeing on which errors are evidence. +// +// Nothing structural prevents that, unlike the WrappedBackend rule in +// hack/lint/ which makes decorator transparency impossible to forget. With two +// callers, both funnelling through one predicate, a ruleguard rule is not worth +// its false positives; if a third appears, it is. Recorded as a phase-3 note. func LastDialErrorOf(b Backend) error { for range maxBackendUnwrapDepth { if b == nil {