From 6b712e76dbce1a683c2c96a17959080cd0fbf850 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Wed, 2 Sep 2026 06:10:26 +0000 Subject: [PATCH] fix(cluster): keep the refusal vocabulary in one table The worker re-classified a failure a local service had already classified. classifyServiceFailure preserved exactly one of the four refusal codes, which was faithful to its own comment for as long as there was one worth keeping; once ErrStreamNotServed existed, a service returning the code whose whole job is to say "I learned nothing" had it promoted to ErrStreamTargetUnavailable, which every reap guard acts on. ErrStreamTagUnknown was promoted too, and cost nothing only because both sides of that one reap. No in-tree service produces either, which is the same "unreachable, therefore safe" argument that let the request-frame merge survive a whole phase, and LocalService is exported. The cause was a fifth site enumerating the vocabulary by hand, so the fix is one table. streamRefusals pairs each sentinel with its wire code and with whether a frontend may act on it as evidence about a backend, and the writer, the reader, IsWorkerAnswer and the new IsStreamRefusal all read it. A fifth code is now taught to every one of them at once. The codes are also pinned against literals written out in a spec, the way this branch already pinned the NATS vocabulary. The round-trip table cannot see a rename, because a rename moves the writer and the reader together; an unrecognised code is deliberately not the worker's answer, so renaming "unavailable" would turn every crashed backend on a tunnelled worker into a row nothing can ever reap, silently and with the suite green. Three comments the previous fix falsified, corrected: - tunnelHeaderTimeout still said the window bounds only framing the frontend writes immediately after opening the stream. That is true on the direct path and false on the relay path, and it was the argument for treating an expiry as the frontend's fault. - classifyServiceFailure's deny-list is three causes, not two: on a dial error net.Error.Timeout also covers ETIMEDOUT and EAGAIN. Both are kept deliberately, because reaping a wedged or resource-starved backend is the eviction this phase exists to prevent, and ECONNREFUSED still reaps. isReadTimeout is renamed reportsTimeout, which is what it asks. - The operator table named three refusals and said a refusal is acted on. It now lists four, with when each is sent and whether the row is reaped. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/services/cluster/dialer.go | 13 +- core/services/cluster/tunnelproto.go | 84 +++++++---- .../services/cluster/tunnelproto_wire_test.go | 130 ++++++++++++++++++ core/services/worker/tunnel.go | 93 +++++++++---- core/services/worker/tunnel_test.go | 42 ++++++ docs/content/features/distributed-mode.md | 17 ++- 6 files changed, 328 insertions(+), 51 deletions(-) create mode 100644 core/services/cluster/tunnelproto_wire_test.go diff --git a/core/services/cluster/dialer.go b/core/services/cluster/dialer.go index 86c6a8bf0..cba17d206 100644 --- a/core/services/cluster/dialer.go +++ b/core/services/cluster/dialer.go @@ -143,9 +143,16 @@ func isAbsenceClaim(err error) bool { // model.transportFailure, whose job is to answer "did this call reach a // backend?" and for whom a refusal means it did. func IsWorkerAnswer(err error) bool { - return errors.Is(err, ErrStreamTagUnknown) || - errors.Is(err, ErrStreamTargetUnavailable) || - errors.Is(err, ErrStreamRequestInvalid) + // Read off the vocabulary table rather than enumerated here, so this + // predicate and the wire codes cannot disagree about which refusals exist. + // Enumerating them by hand is what let a fifth site promote the fourth code + // into a verdict; see streamRefusals. + for _, r := range streamRefusals { + if r.evidence && errors.Is(err, r.sentinel) { + return true + } + } + return false } // dialHandshakeTimeout bounds the request/reply exchange that opens every diff --git a/core/services/cluster/tunnelproto.go b/core/services/cluster/tunnelproto.go index 33daac5db..2e404b247 100644 --- a/core/services/cluster/tunnelproto.go +++ b/core/services/cluster/tunnelproto.go @@ -113,6 +113,53 @@ var ( ErrStreamNotServed = errors.New("cluster: the worker could not serve that stream, for a reason that is not about the backend") ) +// streamRefusals is the whole refusal vocabulary, in ONE table. +// +// The writer, the reader, IsWorkerAnswer and IsStreamRefusal all read it, so a +// fifth refusal cannot be taught to some of them and forgotten in the others. +// That is not hypothetical tidiness: the fourth code was added to the writer, +// the reader and the consumer predicate, and a fifth place that enumerated the +// sentinels by hand (the worker's classifyServiceFailure) silently PROMOTED it +// to a reaping verdict. One table is what makes the next such site impossible +// to write. +// +// ORDER MATTERS for the writer: matching is by errors.Is and the first hit +// wins, so a reason wrapping two sentinels resolves the same way every time. +// +// evidence is the half a consumer acts on, and it is a property OF THE CODE +// rather than of the consumer asking. Three of the four are statements about a +// backend that a frontend may reap on; ErrStreamNotServed is the worker saying +// it learned nothing, and folding it in turns every transient worker-side +// failure into an eviction. See IsWorkerAnswer. +var streamRefusals = []struct { + sentinel error + code string + evidence bool +}{ + {ErrStreamTagUnknown, replyCodeUnknownTag, true}, + {ErrStreamTargetUnavailable, replyCodeUnavailable, true}, + {ErrStreamRequestInvalid, replyCodeBadRequest, true}, + {ErrStreamNotServed, replyCodeNotServed, false}, +} + +// IsStreamRefusal reports whether err ALREADY carries one of this vocabulary's +// classifications. +// +// It answers "has something already decided what this failure is", which is a +// different question from IsWorkerAnswer's "may a consumer act on it". A worker +// that re-classifies an error which already carries a sentinel overwrites a +// decision made closer to the failure, and when the overwrite lands on one of +// the three evidence codes it manufactures a verdict out of something that was +// explicitly not one. +func IsStreamRefusal(err error) bool { + for _, r := range streamRefusals { + if errors.Is(err, r.sentinel) { + return true + } + } + return false +} + // WriteStreamRequest sends the opening frame naming what the stream is for. // // An empty tag is refused here rather than on the wire, because the worker @@ -173,15 +220,11 @@ func WriteStreamAccepted(w io.Writer) error { // distinguished the codes and became a reap-by-omission when one did. func WriteStreamRefusal(w io.Writer, reason error) error { code := replyCodeNotServed - switch { - case errors.Is(reason, ErrStreamTagUnknown): - code = replyCodeUnknownTag - case errors.Is(reason, ErrStreamTargetUnavailable): - code = replyCodeUnavailable - case errors.Is(reason, ErrStreamRequestInvalid): - code = replyCodeBadRequest - case errors.Is(reason, ErrStreamNotServed): - code = replyCodeNotServed + for _, r := range streamRefusals { + if errors.Is(reason, r.sentinel) { + code = r.code + break + } } text := "" @@ -221,21 +264,16 @@ func ReadStreamReply(r io.Reader) error { return fmt.Errorf("reading a tunnel stream reply: unrecognised reply %q", payload) } code, text, _ := strings.Cut(rest, streamRequestSeparator) - switch code { - case replyCodeUnknownTag: - return fmt.Errorf("%w: %s", ErrStreamTagUnknown, text) - case replyCodeUnavailable: - 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 - // retry forever against a refusal that means something else entirely. - return fmt.Errorf("tunnel stream refused with unrecognised code %q: %s", code, text) + for _, r := range streamRefusals { + if code == r.code { + return fmt.Errorf("%w: %s", r.sentinel, text) + } } + // 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 retry + // forever against a refusal that means something else entirely, and so + // IsWorkerAnswer reports false for it and nothing reaps. + return fmt.Errorf("tunnel stream refused with unrecognised code %q: %s", code, text) } // truncateRunes cuts s to at most limit BYTES, on a rune boundary. diff --git a/core/services/cluster/tunnelproto_wire_test.go b/core/services/cluster/tunnelproto_wire_test.go new file mode 100644 index 000000000..0745444da --- /dev/null +++ b/core/services/cluster/tunnelproto_wire_test.go @@ -0,0 +1,130 @@ +// SPDX-License-Identifier: MIT + +package cluster + +// In-package, and that is the point: these specs assert the BYTES a refusal +// puts on the wire, against literals written out here rather than against the +// constants the code uses. A spec that round-trips through this process's own +// writer and reader cannot see a rename, because a rename moves both sides at +// once; the DescribeTable in tunnelproto_test.go is exactly that spec and it +// stays green through any renaming of the four codes. +// +// A wire code is a cross-version contract. A worker and a frontend built from +// different commits talk to each other over it, and the consequence of them +// disagreeing is not a parse error: an unrecognised code is deliberately +// treated as "not the worker's answer", so renaming `unavailable` would turn +// every crashed backend on a tunnelled worker into a row nothing can ever reap, +// silently and with the whole suite green. That is the exact defect this phase +// spent two rounds removing. +// +// This branch set the precedent for pinning a vocabulary against literals in +// core/services/messaging/subjects_wire_test.go, for the same reason. Nothing +// has shipped yet, so no value here is load bearing across a release boundary +// today; that is why the literals may still be changed, and why the change has +// to be deliberate rather than incidental. + +import ( + "bytes" + "encoding/binary" + "errors" + "fmt" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// frameBytes returns the payload of the single frame w wrote, so a spec can +// assert on the bytes rather than on what the reader makes of them. +func frameBytes(write func(w *bytes.Buffer) error) string { + GinkgoHelper() + var buf bytes.Buffer + Expect(write(&buf)).To(Succeed()) + raw := buf.Bytes() + Expect(len(raw)).To(BeNumerically(">=", 2)) + Expect(binary.BigEndian.Uint16(raw[:2])).To(Equal(uint16(len(raw) - 2))) + return string(raw[2:]) +} + +var _ = Describe("the tunnel refusal vocabulary on the wire", func() { + DescribeTable("writes the exact code an older build reads", + func(sentinel error, wantCode string) { + payload := frameBytes(func(w *bytes.Buffer) error { + return WriteStreamRefusal(w, fmt.Errorf("%w: because", sentinel)) + }) + Expect(payload).To(Equal("err " + wantCode + " " + sentinel.Error() + ": because")) + }, + Entry("unknown tag", ErrStreamTagUnknown, "unknown-tag"), + Entry("unavailable target", ErrStreamTargetUnavailable, "unavailable"), + Entry("invalid request", ErrStreamRequestInvalid, "bad-request"), + Entry("nothing learned", ErrStreamNotServed, "not-served"), + ) + + DescribeTable("reads the exact code an older build writes", + func(rawCode string, want error) { + var buf bytes.Buffer + Expect(writeFrame(&buf, "err "+rawCode+" some reason")).To(Succeed()) + Expect(ReadStreamReply(&buf)).To(MatchError(want)) + }, + Entry("unknown tag", "unknown-tag", ErrStreamTagUnknown), + Entry("unavailable target", "unavailable", ErrStreamTargetUnavailable), + Entry("invalid request", "bad-request", ErrStreamRequestInvalid), + Entry("nothing learned", "not-served", ErrStreamNotServed), + ) + + It("accepts a stream with the literal an older build sends", func() { + // The success case has a literal too, and a rename of it would refuse + // every stream rather than mis-classify one, which is at least loud. + Expect(frameBytes(func(w *bytes.Buffer) error { return WriteStreamAccepted(w) })).To(Equal("ok")) + }) + + It("names the two stream tags with the literals the worker routes on", func() { + // The worker's routing table is keyed by these, so a rename here is a + // worker that serves nothing while reporting an unknown tag, which is a + // verdict a frontend acts on. + Expect(StreamTagGRPC).To(Equal("grpc")) + Expect(StreamTagHTTP).To(Equal("http")) + }) + + It("pins which codes a frontend may act on as evidence about a backend", func() { + // The half of the table that decides whether a row is deleted. It is + // asserted against the literals, not against IsWorkerAnswer's own + // output, so moving a code between the two columns reddens here as well + // as at the consumer. + evidence := map[string]bool{} + for _, r := range streamRefusals { + evidence[r.code] = r.evidence + } + Expect(evidence).To(Equal(map[string]bool{ + "unknown-tag": true, + "unavailable": true, + "bad-request": true, + "not-served": false, + }), "a code that changed column changes whether a live model gets evicted") + }) + + It("has exactly one entry per sentinel, and no duplicate codes", func() { + // A duplicate code makes the reader's first match win and the writer's + // first match win, which need not be the same entry. + codes := map[string]int{} + for _, r := range streamRefusals { + Expect(r.sentinel).ToNot(BeNil()) + codes[r.code]++ + } + Expect(codes).To(HaveLen(len(streamRefusals))) + for code, n := range codes { + Expect(n).To(Equal(1), "code %q appears %d times", code, n) + } + }) + + It("classifies every sentinel as a refusal, and nothing else as one", func() { + // IsStreamRefusal is what stops the worker re-classifying a decision + // something closer to the failure already made. Derived from the table + // so a fifth code joins it automatically; asserted here so that + // derivation cannot quietly stop. + for _, r := range streamRefusals { + Expect(IsStreamRefusal(fmt.Errorf("wrapped: %w", r.sentinel))).To(BeTrue(), r.code) + } + Expect(IsStreamRefusal(errors.New("a plain failure"))).To(BeFalse()) + Expect(IsStreamRefusal(nil)).To(BeFalse()) + }) +}) diff --git a/core/services/worker/tunnel.go b/core/services/worker/tunnel.go index cfd33a770..dab799cb0 100644 --- a/core/services/worker/tunnel.go +++ b/core/services/worker/tunnel.go @@ -65,11 +65,24 @@ const ( tunnelHandshakeTimeout = 10 * time.Second // tunnelHeaderTimeout bounds how long a stream may go without sending the - // request frame that says what it is for. It is generous because it bounds - // only the frontend's own framing, which it writes immediately after - // opening the stream; it is present because without it a stream that sends - // nothing holds a goroutine and one of the session's stream slots for as - // long as the tunnel lives. + // request frame that says what it is for. It is present because without it + // a stream that sends nothing holds a goroutine and one of the session's + // stream slots for as long as the tunnel lives. + // + // It is generous because it does NOT bound only the frontend's own framing. + // An earlier version of this comment said it did, which is true on the + // direct path and false on the relay path that carries most of a + // multi-replica deployment's traffic: this 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. A + // whole peer-link round trip therefore runs inside this window, on a link + // deliberately loaded with multi-gigabyte artifacts beside token streams. + // + // The comment mattered because it was the argument for treating an expiry + // as the frontend's fault: framing written immediately can only be late if + // something is wrong with the frontend. It cannot, so an expiry is refused + // with cluster.ErrStreamNotServed and says nothing about a backend. See + // Tunnel.accept. tunnelHeaderTimeout = 15 * time.Second ) @@ -358,7 +371,7 @@ func (t *Tunnel) accept(ctx context.Context, stream net.Conn) (net.Conn, bool) { // 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) { + if reportsTimeout(err) { t.refuse(stream, fmt.Errorf("%w: %v", cluster.ErrStreamNotServed, err)) return nil, false } @@ -408,51 +421,85 @@ func (t *Tunnel) accept(ctx context.Context, stream net.Conn) (net.Conn, bool) { // classifyServiceFailure decides which refusal a local service's error is. // // A service that has ALREADY classified its own failure keeps that -// classification. loopbackService does, and the distinction is not cosmetic: a +// classification, and that is asked of the whole vocabulary rather than of one +// sentinel. loopbackService classifies, and the distinction is not cosmetic: a // target outside this worker's backend port range is a request this worker will // never serve, while a backend that is not listening yet is a condition that // clears on its own. Reporting the first as the second tells a frontend to // retry something that can never work; reporting the second as the first makes // it give up on a backend that is merely starting. // +// This used to preserve ONE of the four codes, which was true to its own +// comment for exactly as long as there was one classification worth keeping. +// Once ErrStreamNotServed existed, a service returning the code whose entire +// job is to say "I learned nothing" had it PROMOTED here into +// ErrStreamTargetUnavailable, which every reap guard acts on. No in-tree +// service produced it, which is the same "unreachable, therefore safe" +// argument that let the request-frame merge survive a whole phase; LocalService +// and TunnelConfig.Services are both exported, so out-of-tree is a real place. +// cluster.IsStreamRefusal reads the vocabulary table, so a fifth code is +// preserved here without anyone remembering to come back. +// // 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. +// which mistake is made. A wrong reap is ACTIVE: it deletes rows and, on the +// inference path, runs ShutdownModel on a model that is loaded and serving. A +// wrong retention is passive, bounded to one replica slot, and clears on a +// restart or an eviction. An allow-list of reapable causes would also fail +// SILENTLY and PERMANENTLY when it missed one, where a deny-list that misses +// fails loudly. So the exemptions below are a DENY-list of causes that are not +// the target answering, 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. +// The exempted causes, stated exactly, because an earlier version of this +// comment said "two" and the predicate covered more than it named: +// +// - The context ending. Here that is the SESSION's context, cancelled while +// stream goroutines are still running, so it means this worker's tunnel is +// being torn down and will reconnect. +// - This process's own I/O deadline (os.ErrDeadlineExceeded). +// - Anything else reporting itself as a net.Error timeout, which on a DIAL +// also covers syscall.ETIMEDOUT and syscall.EAGAIN. Those are kept +// deliberately. EAGAIN is this worker running out of resources, which is +// plainly not about the target. ETIMEDOUT from connect(2) IS an observation +// about the target, but it is the observation "it did not finish the +// handshake", which is a wedged or backlogged listener rather than an +// absent one, and reaping an overloaded backend is the eviction this whole +// phase exists to prevent. ECONNREFUSED, the shape of a process that is +// genuinely gone, is not a timeout and still reaps. +// +// 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) { + if cluster.IsStreamRefusal(err) { return err } - if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || isReadTimeout(err) { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || reportsTimeout(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. +// reportsTimeout reports whether err says of ITSELF that it is a timeout. +// +// Named for what it asks rather than for where it is asked, because it is asked +// in two places that mean different things. On a stream READ it is a deadline +// this process armed. On a DIAL it is wider: Go's syscall.Errno.Timeout is true +// for ETIMEDOUT and EAGAIN as well, and net.OpError passes that through. Both +// call sites want the same ANSWER (not the target speaking, so not evidence +// about a backend), which is why one predicate serves both; see +// classifyServiceFailure for why the wider set is kept deliberately. // // 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 { +func reportsTimeout(err error) bool { if errors.Is(err, os.ErrDeadlineExceeded) { return true } diff --git a/core/services/worker/tunnel_test.go b/core/services/worker/tunnel_test.go index 743cccf72..6afafaa20 100644 --- a/core/services/worker/tunnel_test.go +++ b/core/services/worker/tunnel_test.go @@ -444,6 +444,48 @@ var _ = Describe("Worker tunnel client", func() { Expect(cluster.IsWorkerAnswer(got)).To(BeFalse()) }) + DescribeTable("keeps a classification the local service already made", + // The latent instance of the same shape, found by the gate rather + // than by anything reaching it. This function preserved exactly ONE + // of the four codes, which was faithful to its own comment for as + // long as there was one worth keeping. Once ErrStreamNotServed + // existed, a service returning the code whose whole job is to say + // "I learned nothing" had it PROMOTED to ErrStreamTargetUnavailable, + // which every reap guard acts on. + // + // No in-tree service produced it, which is the "unreachable, + // therefore safe" argument that let the request-frame merge survive + // a whole phase. LocalService and TunnelConfig.Services are both + // exported, so out-of-tree is a real place, and the same standard + // applies. + func(classified error, wantEvidence bool) { + frontend = newFakeFrontend(false) + start(func(c *TunnelConfig) { + c.Services[cluster.StreamTagGRPC] = func(context.Context, string) (net.Conn, error) { + return nil, fmt.Errorf("the service decided for itself: %w", classified) + } + }) + + 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(classified), + "re-classifying overwrites a decision made closer to the failure") + Expect(cluster.IsWorkerAnswer(got)).To(Equal(wantEvidence)) + }, + // The one the promotion broke: not evidence before, evidence after. + Entry("I learned nothing", cluster.ErrStreamNotServed, false), + // Promoted too. Both sides reap, so it cost nothing, which is + // exactly why nothing caught it. + Entry("I do not serve that tag", cluster.ErrStreamTagUnknown, true), + Entry("that request was malformed", cluster.ErrStreamRequestInvalid, true), + Entry("I could not reach the target", cluster.ErrStreamTargetUnavailable, true), + ) + 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 diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index 636ecde49..9075c97a4 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -188,11 +188,24 @@ These outcomes are kept apart on purpose, because they call for different action | Not the owner | The routing was stale | Resolve the owner again | | Peer unreachable | A replica exists and will not answer | Retry | | No relay path | This replica cannot reach the owner at all | Report; requests here fail until it can | -| The worker refused | The worker answered and said no | Report, and **act on it**; the worker is connected and speaking about its own backend | +| The worker refused | The worker answered and said no | Depends on WHICH refusal; see below | **None of the first four is absence.** A worker's presence is its **heartbeat**, and a route to it is a separate fact that can be false while the worker is registered, heartbeating and serving every request another replica sends it. So the frontend answers "no route", never "this worker is gone", and none of the first four causes a model to be rescheduled or a `node_models` row to be deleted. -The fifth is different, and deliberately so. A worker that **refuses** a stream has answered, which proves it is connected; what it is refusing is the stream to one backend process on it. That is the ordinary shape of a crashed backend now that workers listen on nothing: the worker's own dial to the process fails and it says so. The frontend treats that as evidence about the backend, so the model's row is reaped and the replica is reloaded, exactly as a dead local backend would be. Without that, a crashed backend on a healthy worker would leave a row nothing could ever delete, its replica slot permanently occupied. A refusal code the frontend does not recognise - a newer worker's vocabulary - is treated as "no route" instead, so a version skew costs a retry rather than a reaped replica. +The fifth is different, and deliberately so. A worker that **refuses** a stream has answered, which proves it is connected; what it is refusing is the stream to one backend process on it. That is the ordinary shape of a crashed backend now that workers listen on nothing: the worker's own dial to the process fails and it says so. + +There are **four** refusals, and only three of them are evidence about a backend. The distinction decides whether a model's row is deleted, so an operator reading one of these in a log can tell what will happen next: + +| Refusal a worker sends | When | Row reaped? | +|---|---|---| +| `the worker could not reach the local service for that stream` | The worker's own dial to the backend process was refused. A crashed backend | **Yes.** Reloaded elsewhere, as a dead local backend would be | +| `the worker does not serve that stream tag` | The worker does not serve that kind of stream at all | **Yes.** Nothing clears this until the worker is upgraded, and the model re-registers somewhere that works | +| `the worker rejected the stream request as malformed` | The stored backend address is not a port in this worker's range | **Yes.** The row can never be reached, so reaping lets the model re-register a usable address | +| `the worker could not serve that stream, for a reason that is not about the backend` | The request frame did not arrive in the worker's 15s window, the worker's tunnel was being torn down, or it ran out of a local resource | **No.** These clear on their own; the request fails with "no route" and is retried | + +The fourth exists because the other three are acted on. A relayed request crosses a peer link before its frame reaches the worker, so on a congested link a frame can arrive late through nobody's fault; reported as one of the first three, that would evict a model that is loaded and serving. If you see the fourth in your logs, look at peer-link congestion or a worker that is reconnecting, not at the backend it names. + +A refusal code the frontend does not recognise - a newer worker's vocabulary - is treated as "no route" as well, so a version skew costs a retry rather than a reaped replica. That distinction is the whole point rather than a nicety. A scheduler told that a connected worker has gone away stops its backend and reclaims every model it is running, and the events that produce "no route" are ordinary ones: a frontend replica restarting, an ownership row a moment stale, a worker that has not dialled its tunnel yet. A worker is treated as absent only when its **heartbeat** goes stale, which is a separate mechanism with its own threshold (see `--stale-node-threshold`).