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