From 5880f4e6dded32e1a35c1dc49f2bb3bedf8fef47 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Wed, 2 Sep 2026 20:45:01 +0000 Subject: [PATCH] test(distributed): pin the no-demotion rule at every call site it is stated Review fix round 1. Two blocking findings and seven non-blocking; both blocking ones are M12's shape again, and this time on the invariant itself. No production behaviour changes here: everything below was already correct and merely unpinned, so re-inserting the defect left all 679 specs green. The only non-comment edits are one struct-field comment and one log message. "A failed control RPC no longer demotes a node" is stated three times in this package and was pinned once, at ListBackends. Putting MarkUnhealthy back at either op-drain site passed. What that buys in production is the fleet-wide eviction this phase exists to prevent: MarkUnhealthy removes a node from ListDuePendingBackendOps AND from scheduling, so a frontend replica that has just lost its tunnels demotes every node it holds an op for, for a reason that is about the frontend. The reconciler's is the worse of the two, being a background loop nobody is watching. Both now have a spec, each with the recorded op failure as its negative control so "still healthy" cannot pass by nothing having happened. The sweep the review asked for found four more rules stated at more call sites than they were pinned at, and two the review had not: The still-installing surfacing at the manager layer has two call sites and was pinned at InstallBackend. Dropping it from UpgradeBackend reported a spent budget as GREEN SUCCESS: the admin sees the upgrade finished while the worker is still re-pulling gigabytes. The agent-node skip has two call sites and was pinned at ListBackends. Without it the fan-out enqueues a row for every agent node, and an agent worker serves no control plane, so that row can never drain: it retries until the dead-letter cap. The still-installing conversion has three call sites and was pinned at two; the legacy force-install fallback was the gap. Its budget was unpinned too, so the new spec asserts both, on the upgrade budget rather than the install one, since the fallback re-fires an install as part of an upgrade. The carrier split has two call sites and was pinned at one. Hardcoding NodeTypeBackend in UnloadRemoteModelContext passed, and an agent node holding a node_models row would then have its stop sent over a tunnel it does not hold, fail, and leave the row behind. The new spec unloads a model held by one node of each kind and asserts each stop went to that node's own carrier and to no other. router_nats_liveness_test.go asserted demote-on-absence, which production can no longer produce, and its header described the pre-cutover world. The exclusion is unreachable by construction rather than by argument: cluster, the package supplying every control-path dial error, does not link nats.go at all. The file now says that, and gains the assertion that IS load-bearing, a table naming each sentinel a control RPC can answer with and requiring that none of them excludes. Widening the exclusion to ErrWorkerUnroutable reddens four of its entries plus the real-adapter scheduling spec. unroutable keeps no budget-first guard and the reason is now written at it: unlike controlFailure it reads one already-recorded error rather than racing a live deadline, and an expiry is not in streamRefusals, so it falls to the umbrella without one. The two implement the same split at two layers and each now names the other. Fourteen comments still described the bus. Among them the reconciler saying a drain would "churn NATS every tick", a spec comment naming a subject builder this branch deleted, and the agent-skip comment explaining the skip by a subscription that no longer exists. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/services/nodes/interfaces.go | 6 ++ core/services/nodes/managers_distributed.go | 31 ++++--- .../nodes/managers_distributed_test.go | 90 ++++++++++++++++++- core/services/nodes/reconciler.go | 15 ++-- core/services/nodes/reconciler_test.go | 23 +++++ core/services/nodes/router_liveness.go | 9 +- .../nodes/router_nats_liveness_test.go | 58 ++++++++++-- core/services/nodes/unloader.go | 5 +- core/services/nodes/unloader_test.go | 64 +++++++++++++ 9 files changed, 267 insertions(+), 34 deletions(-) diff --git a/core/services/nodes/interfaces.go b/core/services/nodes/interfaces.go index 6a0a1367c..bce46dc33 100644 --- a/core/services/nodes/interfaces.go +++ b/core/services/nodes/interfaces.go @@ -255,6 +255,12 @@ var ErrNoWorkerDialer = fmt.Errorf("%w: no worker tunnel dialer is configured", // 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. +// The control plane's sibling is nodes.controlFailure, which splits the same +// two ways. That one checks the caller's remaining budget FIRST, because it +// races a live deadline against a reply that may arrive in the same instant. +// There is no such race here: this reads ONE error that was already recorded on +// the client, and an expiry is not in streamRefusals, so it falls to the +// umbrella below without a guard. Anyone changing the split must change both. 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/managers_distributed.go b/core/services/nodes/managers_distributed.go index e3b6ae5b5..2db66e9d0 100644 --- a/core/services/nodes/managers_distributed.go +++ b/core/services/nodes/managers_distributed.go @@ -16,8 +16,9 @@ import ( "github.com/mudler/xlog" ) -// DistributedModelManager wraps a local ModelManager and adds NATS fan-out -// for model deletion so worker nodes clean up stale files. +// DistributedModelManager wraps a local ModelManager and fans model deletion +// out to the worker nodes so they clean up stale files. The fan-out is a +// control RPC over each worker's tunnel; see RemoteUnloaderAdapter. type DistributedModelManager struct { local galleryop.ModelManager adapter *RemoteUnloaderAdapter @@ -55,8 +56,9 @@ type nodeProgressSink interface { UpdateNodeProgress(opID, nodeID string, np galleryop.NodeProgress) } -// DistributedBackendManager wraps a local BackendManager and adds NATS fan-out -// for backend deletion so worker nodes clean up stale files. +// DistributedBackendManager wraps a local BackendManager and fans backend +// deletion out to the worker nodes so they clean up stale files. The fan-out is +// a control RPC over each worker's tunnel; see RemoteUnloaderAdapter. type DistributedBackendManager struct { local galleryop.BackendManager adapter *RemoteUnloaderAdapter @@ -121,7 +123,7 @@ func (r BackendOpResult) Err() error { // nodes get an immediate attempt; success deletes the row, failure records // the error and leaves the row for the reconciler to retry. // -// `apply` is the NATS round-trip for one node. Returning an error keeps the +// `apply` is the control RPC for one node. Returning an error keeps the // row in the queue and marks the per-node status as "error"; returning nil // deletes the row and reports "success". For non-healthy nodes the status // is "queued" — no attempt is made right now, reconciler will pick it up @@ -167,7 +169,7 @@ func (d *DistributedBackendManager) enqueueAndDrainBackendOp(ctx context.Context continue } // Backend lifecycle ops only make sense on backend-type workers. - // Agent workers don't subscribe to backend.install/delete/list, so + // Agent workers hold no tunnel and serve no control plane, so // enqueueing for them guarantees a forever-retrying row that the // reconciler can never drain. Silently skip - they aren't consumers. if node.NodeType != "" && node.NodeType != NodeTypeBackend { @@ -454,7 +456,7 @@ func (d *DistributedBackendManager) clearSatisfiedInstallRows(ctx context.Contex // InstallBackend fans out installation through the pending-ops queue so // non-healthy nodes get retried when they come back instead of being silently -// skipped. Reply success from the NATS round-trip deletes the queue row; +// skipped. Reply success from the control RPC deletes the queue row; // reply.Success==false is treated as an error so the row stays for retry. // // When op.TargetNodeID is set, only that node is visited - the same allowlist @@ -494,9 +496,10 @@ func (d *DistributedBackendManager) InstallBackend(ctx context.Context, op *gall }) } } - // nil-callback shortcut: when there is nothing to deliver to, - // hand the adapter a nil onProgress so it skips the per-op NATS - // subscription. Matches the pre-Phase-4 bridgeProgressCb semantics. + // nil-callback shortcut: when there is nothing to deliver to, hand the + // adapter a nil onProgress so it discards the worker's progress lines + // instead of decoding them. They ride the install response itself, so + // there is nothing to arrange either way. var onProgressArg func(messaging.BackendInstallProgressEvent) if progressCb != nil || d.progressSink != nil { onProgressArg = onProgress @@ -531,7 +534,7 @@ func (d *DistributedBackendManager) InstallBackend(ctx context.Context, op *gall return nil } -// UpgradeBackend uses a separate NATS subject (backend.upgrade) so the slow +// UpgradeBackend uses a separate control verb (backend.upgrade) so the slow // force-reinstall path doesn't head-of-line-block routine model loads on // the same worker. Only nodes that already report this backend as installed // are targeted — fanning out to every node would ask workers to "upgrade" @@ -629,8 +632,10 @@ func (d *DistributedBackendManager) UpgradeBackend(ctx context.Context, op *gall return hardErr } // Same in-progress surfacing as InstallBackend: a long-running worker - // upgrade that timed out the NATS round-trip must not be reported as - // green success. + // upgrade that outlived the caller's budget must not be reported as green + // success. Pinned by "reports an upgrade that ran out of budget as still + // installing", because this is the second of the rule's two call sites and + // dropping it here left the suite green. for _, n := range result.Nodes { if n.Status == galleryop.NodeStatusRunningOnWorker { return fmt.Errorf("%w: %s", galleryop.ErrWorkerStillInstalling, summarizeRunningOnWorker(result.Nodes)) diff --git a/core/services/nodes/managers_distributed_test.go b/core/services/nodes/managers_distributed_test.go index 62abd3fc9..68b8b6420 100644 --- a/core/services/nodes/managers_distributed_test.go +++ b/core/services/nodes/managers_distributed_test.go @@ -318,6 +318,67 @@ var _ = Describe("DistributedBackendManager", func() { }) }) + // The agent skip is stated at TWO call sites, the fan-out and + // ListBackends, and was pinned only at ListBackends. Agent workers + // serve no control plane, so a row enqueued for one can never be + // drained: it retries every reconciler tick until the dead-letter cap + // and shows in the UI as an operation that never finishes. + It("enqueues nothing for an agent node, which serves no control plane", func() { + backend := registerHealthyBackend("worker-a", "10.0.0.1:50051") + agent := &BackendNode{Name: "agent-a", NodeType: NodeTypeAgent, Address: "10.0.0.2:50051"} + Expect(registry.Register(ctx, agent, true)).To(Succeed()) + + mc.scriptReply(controlKey(backend.ID, workerctl.PathBackendInstall), + messaging.BackendInstallReply{Success: true, WorkerLocalAddress: "10.0.0.1:50100"}) + + Expect(mgr.InstallBackend(ctx, op("vllm-development"), nil)).To(Succeed()) + + // The backend node WAS asked, which is the negative control: a + // fan-out that skipped everything would satisfy the agent + // assertion by doing nothing at all. + Expect(mc.callSubjects()).To(Equal([]string{controlKey(backend.ID, workerctl.PathBackendInstall)})) + rows, err := registry.ListPendingBackendOps(ctx) + Expect(err).ToNot(HaveOccurred()) + for _, row := range rows { + Expect(row.NodeID).ToNot(Equal(agent.ID), + "an agent node cannot drain a backend op, so it must never be given one") + } + }) + + // The admin fan-out is the third call site of the rule that a failed + // control RPC does not demote a node, and it was the unpinned one. The + // rule is stated in three places in this package and, before this spec, + // pinned only at ListBackends. + // + // What the demotion buys in production is the fleet-wide eviction this + // phase exists to prevent: MarkUnhealthy removes the node from + // ListDuePendingBackendOps AND from scheduling, so a frontend replica + // re-homing its tunnels would take out every node it has an op for. + Context("when the install RPC could not be routed", func() { + It("records the failure without demoting the node", func() { + node := registerHealthyBackend("worker-unroutable", "10.0.0.8:50051") + mc.scriptUnroutable(node.ID) + + err := mgr.InstallBackend(ctx, op("vllm"), nil) + Expect(err).To(HaveOccurred()) + // Not the still-installing soft path: that branch returns before + // the failure handling this spec is about, so without this the + // assertions below could pass on the wrong branch. + Expect(errors.Is(err, galleryop.ErrWorkerStillInstalling)).To(BeFalse()) + + // The recorded failure witnesses that the fan-out ran and + // reached the branch that used to demote. + rows, listErr := registry.ListPendingBackendOps(ctx) + Expect(listErr).ToNot(HaveOccurred()) + Expect(rows).To(HaveLen(1)) + Expect(rows[0].Attempts).To(Equal(1)) + + after, getErr := registry.Get(ctx, node.ID) + Expect(getErr).ToNot(HaveOccurred()) + Expect(after.Status).To(Equal(StatusHealthy)) + }) + }) + Context("ListBackends clears confirmed install rows", func() { It("deletes the pending_backend_ops install row when the backend is reported installed on its target node", func() { node := registerHealthyBackend("worker-a", "10.0.0.5:50051") @@ -573,7 +634,7 @@ var _ = Describe("DistributedBackendManager", func() { scriptNoBackends(lacks.ID) mc.scriptReply(controlKey(has.ID, workerctl.PathBackendUpgrade), messaging.BackendUpgradeReply{Success: true}) - // Deliberately don't script SubjectNodeBackendUpgrade for `lacks`: + // Deliberately don't script backend.upgrade for `lacks`: // if the manager attempts it, the scripted-client default returns // fakeNoRespondersErr and the assertion below fails loudly. @@ -667,6 +728,33 @@ var _ = Describe("DistributedBackendManager", func() { }) }) + // The still-installing surfacing has TWO call sites, InstallBackend and + // this one, and was pinned only at InstallBackend. Dropping it here + // reports a spent budget as GREEN SUCCESS: the admin sees the upgrade + // finished while the worker is still re-pulling gigabytes, and the row + // the reconciler needs to confirm it is invisible in the UI. + It("reports an upgrade that ran out of budget as still installing, not as success", func() { + n := registerHealthyBackend("worker-slow", "10.0.0.1:50051") + scriptInstalled("vllm-development", n.ID) + // Only the upgrade verb hangs: backend.list must still answer, or + // the manager never gets as far as the node it would upgrade. + mc.scriptHang(controlKey(n.ID, workerctl.PathBackendUpgrade)) + slow := &DistributedBackendManager{ + local: stubLocalBackendManager{}, + adapter: NewRemoteUnloaderAdapter(nil, nil, mc.controlClient(), time.Minute, 200*time.Millisecond), + registry: registry, + } + + err := slow.UpgradeBackend(ctx, upgradeOp("vllm-development"), nil) + Expect(err).To(HaveOccurred()) + Expect(errors.Is(err, galleryop.ErrWorkerStillInstalling)).To(BeTrue(), + "a spent budget must not read as a finished upgrade, got %v", err) + + rows, listErr := registry.ListPendingBackendOps(ctx) + Expect(listErr).ToNot(HaveOccurred()) + Expect(rows).To(HaveLen(1), "the row the reconciler confirms the outcome from must survive") + }) + // Rolling-update fallback: pre-2026-05-08 workers do not serve // backend.upgrade, so the manager catches the worker's own 404 and // re-fires the legacy backend.install Force=true on the same node. diff --git a/core/services/nodes/reconciler.go b/core/services/nodes/reconciler.go index ed231d93e..cdefbc666 100644 --- a/core/services/nodes/reconciler.go +++ b/core/services/nodes/reconciler.go @@ -57,7 +57,7 @@ type ModelProber interface { } // NodeProcessLister asks a worker which model backend processes it currently -// has running. Implemented by RemoteUnloaderAdapter over NATS. +// has running. Implemented by RemoteUnloaderAdapter over the worker's tunnel. // // This is the sounder liveness signal: the worker owns the process table, so // its answer does not depend on whether a backend is busy. A health probe @@ -161,7 +161,7 @@ type ReplicaReconciler struct { registry *NodeRegistry scheduler ModelScheduler // interface for scheduling new models unloader NodeCommandSender - adapter *RemoteUnloaderAdapter // NATS sender for pending-op drain + adapter *RemoteUnloaderAdapter // control-RPC sender for the pending-op drain prober ModelProber // health probe for model gRPC addrs db *gorm.DB interval time.Duration @@ -209,7 +209,7 @@ type ReplicaReconcilerOptions struct { Registry *NodeRegistry Scheduler ModelScheduler Unloader NodeCommandSender - // Adapter is the NATS sender used to retry pending backend ops. When nil, + // Adapter is the control-RPC sender used to retry pending backend ops. When nil, // the state-reconciler pending-drain pass is a no-op (single-node mode). Adapter *RemoteUnloaderAdapter // RegistrationToken is the bearer token the default gRPC prober presents to @@ -442,8 +442,9 @@ func (rc *ReplicaReconciler) drainPendingBackendOps(ctx context.Context) { // Dead-letter cap: after maxAttempts the row is the reconciler // equivalent of a poison message. Delete it loudly so the queue - // doesn't churn NATS every tick forever — operators can re-issue - // the op from the UI if they still want it applied. + // doesn't churn a control RPC at the worker every tick forever — + // operators can re-issue the op from the UI if they still want it + // applied. if op.Attempts+1 >= maxPendingBackendOpAttempts { xlog.Error("Reconciler: abandoning pending backend op after max attempts", "op", op.Op, "backend", op.Backend, "node", op.NodeID, @@ -692,8 +693,8 @@ const workerMissesBeforeReap = 2 // from ever being mistaken for a dead one. // // A worker that cannot be reached is skipped rather than treated as empty. A -// messaging failure says nothing about the processes, and assuming the worst -// would delete a whole node's rows on a transient NATS blip; the port probe +// failure to route says nothing about the processes, and assuming the worst +// would delete a whole node's rows every time a tunnel re-homed; the port probe // remains as the fallback for those nodes. func (rc *ReplicaReconciler) reconcileNodeProcesses(ctx context.Context) { if rc.processLister == nil { diff --git a/core/services/nodes/reconciler_test.go b/core/services/nodes/reconciler_test.go index 075fee25f..64171d185 100644 --- a/core/services/nodes/reconciler_test.go +++ b/core/services/nodes/reconciler_test.go @@ -947,6 +947,29 @@ var _ = Describe("ReplicaReconciler — state reconciliation", func() { Expect(rows[0].Attempts).To(Equal(1)) }) + // A failed op must not DEMOTE the node, and this is the second of the + // rule's three call sites. Marking unhealthy here takes the node out of + // ListDuePendingBackendOps and out of scheduling at the same time, so a + // frontend replica that has just lost its tunnels would evict every node + // with a queued op, fleet-wide, for a reason that is about the frontend. + // + // The surviving row is the negative control: it witnesses that the drain + // actually ran and actually failed, so "still healthy" cannot pass by + // nothing having happened. + It("does NOT demote a node whose queued op it could not route", func() { + workers.scriptUnroutable(node.ID) + + rc.drainPendingBackendOps(context.Background()) + + rows := queuedOps() + Expect(rows).To(HaveLen(1), "the drain must have run and failed") + Expect(rows[0].Attempts).To(Equal(1)) + + after, err := registry.Get(context.Background(), node.ID) + Expect(err).ToNot(HaveOccurred()) + Expect(after.Status).To(Equal(StatusHealthy)) + }) + // The same rule for the other non-answer: no route at all. The install // is unreachable too here, so the witness is the error the drain // RECORDED, which names the verb that actually failed. diff --git a/core/services/nodes/router_liveness.go b/core/services/nodes/router_liveness.go index 9809d1b2d..38c078384 100644 --- a/core/services/nodes/router_liveness.go +++ b/core/services/nodes/router_liveness.go @@ -41,8 +41,11 @@ func (r *SmartRouter) nodeAnswersOnBus(node *BackendNode) bool { return !errors.Is(err, nats.ErrNoResponders) } -// pickReachableNode calls selectNode until it yields a node that still answers -// on the bus, and returns nil when it cannot find one. +// pickReachableNode calls selectNode until it yields a node that nodeAnswersOnBus +// does not exclude, and returns nil when it cannot find one. +// +// No control-RPC outcome excludes, so today this returns the first node the +// selector offers. The loop stays until the scheduler reads cluster.Presence. // // A node that does not answer is marked unhealthy before the next attempt. That // both removes it from the next selection, which queries only healthy nodes, @@ -57,7 +60,7 @@ func (r *SmartRouter) pickReachableNode(ctx context.Context, selectNode func() * if r.nodeAnswersOnBus(node) { return node } - xlog.Warn("Scheduled node is not answering on the bus, marking unhealthy and re-scheduling", + xlog.Warn("Scheduled node was excluded by the liveness probe, marking unhealthy and re-scheduling", "node", node.Name, "nodeID", node.ID) if err := r.registry.MarkUnhealthy(ctx, node.ID); err != nil { // Without the demotion the next selection would hand back the same diff --git a/core/services/nodes/router_nats_liveness_test.go b/core/services/nodes/router_nats_liveness_test.go index ec4820c9f..d35d5a088 100644 --- a/core/services/nodes/router_nats_liveness_test.go +++ b/core/services/nodes/router_nats_liveness_test.go @@ -3,17 +3,34 @@ package nodes import ( "context" "errors" + "fmt" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/core/services/cluster" ) -// A node's stored status comes from its HTTP heartbeat, but work is dispatched -// over NATS. A worker that dies stops answering on the bus immediately and -// keeps its healthy status until the heartbeat ages out, so the scheduler could -// commit to a node it could not reach. The request then failed outright with -// "no responders available" rather than moving to a node that was actually up. -var _ = Describe("Scheduling past a node that left the bus", func() { +// pickReachableNode's exclusion branch, which no longer runs in production. +// +// It excludes on nats.ErrNoResponders alone, and since the control plane moved +// onto the tunnel nothing can produce that value: cluster, the package that +// supplies every control-path dial error, does not link nats.go at all. So the +// specs below that drive fakeUnloader.deadNodes exercise a branch a real +// adapter cannot enter. They are kept, rather than deleted, because the LOOP is +// still compiled and still executed on every scheduling decision, and its shape +// (the retry bound, and stopping when the demotion itself fails) is what stops a +// future re-wiring from spinning. They go with the exclusion in the task that +// gives the scheduler cluster.Presence. +// +// What is load-bearing TODAY is the opposite assertion, and it lives in two +// places: the DescribeTable at the bottom of this file, which pins that none of +// the sentinels a control RPC actually produces excludes anything, and the +// scheduling specs in unloader_ping_test.go, which prove the same thing through +// a REAL RemoteUnloaderAdapter. This file's doubles cannot see a carrier +// change; that is why they stayed green through the window in which every +// healthy worker read as absent. +var _ = Describe("The retired bus-absence exclusion", func() { var ( reg *fakeModelRouter fake *fakeUnloader @@ -96,8 +113,8 @@ var _ = Describe("Scheduling past a node that left the bus", func() { Expect(fake.pingCalls).To(Equal([]string{"dead-node"})) }) - // Only a no-responders answer proves absence. Excluding a node that is - // merely slow would cost real capacity. + // The half of the branch that IS still reachable: every error a control + // RPC can produce lands here. It("keeps a node that answers slowly or errors for another reason", func() { slow := newNode("slow-node") fake.pingErr = errors.New("timeout waiting for reply") @@ -115,4 +132,29 @@ var _ = Describe("Scheduling past a node that left the bus", func() { Expect(plain.pickReachableNode(context.Background(), selectorReturning(node))).To(Equal(node)) }) + + // The rule that actually holds now, stated in the vocabulary a control RPC + // answers in. Widening the exclusion to any of these puts the scheduler + // back in the state this task was opened to fix: a worker that is + // heartbeating and serving, unroutable from THIS replica for a moment + // while its tunnel re-homes, demoted for every scheduler in the cluster. + // + // unloader_ping_test.go proves the same thing through a real adapter and a + // real transport. This table is the cheap negative control beside it: it + // names each sentinel, so a widening is attributed to the sentinel that + // caused it rather than to "something in the ping". + DescribeTable("never excludes a node on anything a control RPC can answer with", + func(cause error) { + fake.pingErr = cause + node := newNode("live-node") + + Expect(router.pickReachableNode(context.Background(), selectorReturning(node))).To(Equal(node)) + Expect(reg.markedUnhealthy).To(BeEmpty()) + }, + Entry("no route to the worker", fmt.Errorf("control rpc: %w: %w", ErrWorkerUnroutable, cluster.ErrNoRoute)), + Entry("no tunnel dialer wired", ErrNoWorkerDialer), + Entry("the worker is older than this frontend", ErrWorkerControlUnsupported), + Entry("the caller's budget ran out", fmt.Errorf("control rpc: %w: %w", ErrWorkerUnroutable, context.DeadlineExceeded)), + Entry("the worker's own refusal", fmt.Errorf("%w: no such process", cluster.ErrStreamTargetUnavailable)), + ) }) diff --git a/core/services/nodes/unloader.go b/core/services/nodes/unloader.go index 542e7d8bd..be675ed2b 100644 --- a/core/services/nodes/unloader.go +++ b/core/services/nodes/unloader.go @@ -133,8 +133,9 @@ func (a *RemoteUnloaderAdapter) StopModelReplica(ctx context.Context, nodeID str return reply, nil } -// UnloadRemoteModel finds the node(s) hosting the given model and tells them -// to stop their backend process via NATS backend.stop event. +// UnloadRemoteModel finds the node(s) hosting the given model and tells each +// to stop its backend process. The carrier is decided per node by its type, +// which is why stopBackend takes one: see stopBackend. // The worker process handles a bounded Free() followed by process termination; // forced shutdown skips Free(). // This is called by ModelLoader.deleteProcess() when process == nil (remote model). diff --git a/core/services/nodes/unloader_test.go b/core/services/nodes/unloader_test.go index 9ea61e670..d276b16ae 100644 --- a/core/services/nodes/unloader_test.go +++ b/core/services/nodes/unloader_test.go @@ -274,6 +274,32 @@ var _ = Describe("RemoteUnloaderAdapter", func() { Expect(json.Unmarshal(workers.calls[0].Data, &payload)).To(Succeed()) Expect(payload).To(Equal(messaging.BackendStopRequest{Backend: "llama", Force: true})) }) + + // The carrier split has TWO call sites, which is why stopBackend takes + // nodeType as a parameter. StopBackend is pinned in both directions + // below; this is the other caller, and hardcoding NodeTypeBackend here + // used to leave the whole suite green. An agent node holding a + // node_models row would then have its stop sent over a tunnel it does + // not hold, the call would fail, and the replica row would be left + // behind. + It("routes each stop by ITS node's type, not by one choice for the unload", func() { + locator.nodes = []BackendNode{ + {ID: "agent-1", Name: "agent", NodeType: NodeTypeAgent}, + {ID: "backend-1", Name: "gpu-1", NodeType: NodeTypeBackend}, + } + scriptStop("backend-1") + + Expect(adapter.UnloadRemoteModelContext(context.Background(), "llama", false)).To(Succeed()) + + // The agent's stop went to the bus and only the agent's did; the + // backend's went over the tunnel and only the backend's did. + Expect(bus.publishedSubjects()).To(Equal([]string{messaging.SubjectNodeBackendStop("agent-1")})) + Expect(workers.callSubjects()).To(Equal([]string{controlKey("backend-1", workerctl.PathBackendStop)})) + // Both rows dropped, which is the negative control: a stop put on + // the carrier the other kind of worker listens on fails, and a + // failed stop keeps its row. + Expect(locator.removedPairs).To(HaveLen(2)) + }) }) // The carrier split. It is the one verb of the ten that is decided by the @@ -500,6 +526,44 @@ var _ = Describe("RemoteUnloaderAdapter timeout handling", func() { Expect(errors.Is(err, ErrWorkerUnroutable)).To(BeTrue()) }) + // The still-installing rule has THREE call sites: InstallBackend, + // UpgradeBackend and the legacy force-install fallback. The first two are + // pinned by the specs above and by the timeout-configuration pair; the + // fallback was the unpinned one, and widening its guard to any error left + // the whole suite green. A permanently unroutable worker on the fallback + // path would then report as still-installing forever, which galleryop + // treats as a soft failure: the retry is pushed out and the operator never + // sees the error. + It("reports a spent budget on the legacy fallback as still-installing, on the upgrade budget", func() { + workers := newScriptedControlWorkers() + workers.scriptHang(controlKey("n1", workerctl.PathBackendInstall)) + // Install and upgrade budgets far apart: the fallback re-fires an + // INSTALL but is part of an upgrade, so it must wait the upgrade + // budget. Waiting the install one would satisfy a bare + // still-installing assertion while carrying the wrong deadline. + adapter := NewRemoteUnloaderAdapter(nil, nil, workers.controlClient(), 100*time.Millisecond, 500*time.Millisecond) + + started := time.Now() + _, err := adapter.installWithForceFallback("n1", "llama-cpp", "[]", "", "", "", 0, "", nil) + elapsed := time.Since(started) + + Expect(errors.Is(err, galleryop.ErrWorkerStillInstalling)).To(BeTrue(), + "a spent budget is reported as still-installing, got %v", err) + Expect(elapsed).To(BeNumerically(">=", 500*time.Millisecond)) + }) + + It("does NOT report an unroutable legacy fallback as still installing", func() { + workers := newScriptedControlWorkers() + // The verb is not scripted, so the worker fails to SERVE it rather + // than answering; that is a 5xx and lands under the no-route umbrella. + adapter := NewRemoteUnloaderAdapter(nil, nil, workers.controlClient(), time.Minute, time.Minute) + + _, err := adapter.installWithForceFallback("n1", "llama-cpp", "[]", "", "", "", 0, "", nil) + Expect(err).To(HaveOccurred()) + Expect(errors.Is(err, galleryop.ErrWorkerStillInstalling)).To(BeFalse()) + Expect(errors.Is(err, ErrWorkerUnroutable)).To(BeTrue()) + }) + It("does not read a message merely containing the words nats timeout as a timeout", func() { // The string match the bus carrier needed would have matched a worker // error that quoted the phrase, and would have turned a real failure