Cluster upgrades: nodes told together with progress said, done when back on the version, the download kept past a broken stream

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -48,6 +48,18 @@ APIs change between minors without a deprecation period.
 
 ### Fixed
 
+- **A cluster upgrade that sat silent, or stopped on a node that had upgraded.**
+  - The nodes of a wave are told together, each saying when it is done, and what is still awaited is said every 30
+    seconds; before, each was told in turn, up to five minutes each, with nothing on the screen.
+  - The control plane takes the node registering again on the version as done: the agent's answer, sent a moment
+    before it exits, could be lost, and a node that had upgraded was failed as "older than v0.10.4".
+  - The agent's download no longer ends with its connection (a control plane restarting, leadership moving), and the
+    control plane waits longer than the download may take; a second request while one runs is refused.
+  - A cordoned or draining node upgraded is done; a node whose idle model stopped meanwhile is done once nothing on it
+    is starting; a node not connected is passed over and named at the end rather than stopping every node after it.
+  - The console's node upgrade waits as long as the control plane, and offers no development version.
+  - systemd never gives up restarting the agent (`StartLimitIntervalSec=0`), in units written from now on.
+
 - **Strata.** One named by hand in `agent.yaml` was never reported, so never placed on; its engine, which leaves the
   server's process group, is now stopped with it; Pascal and Volta, which its CUDA 12 build runs, were refused.
 - **A name with a slash in it, asked through a standby control plane.** `llmgw ctl kb delete Demo/hr` on a standby

deploy/systemd/llmgw-agent.service

@@ -2,6 +2,9 @@
 Description=llmgw node agent
 After=network-online.target
 Wants=network-online.target
+# Never given up on: a new release that fails its first starts, or an upgrade
+# that restarts it twice in a minute, would otherwise leave the node off.
+StartLimitIntervalSec=0
 
 [Service]
 ExecStart=/usr/local/bin/llmgw agent --config /etc/llmgw/agent.yaml

internal/agent/agent.go

@@ -47,7 +47,9 @@ import (
 
 // Agent runs on a GPU node.
 type Agent struct {
-	cfg Config
+	// upgrading is true while an upgrade downloads.
+	upgrading atomic.Bool
+	cfg       Config
 	// answered says the current session heard from its control plane, which
 	// starts the reconnect backoff again when it ends.
 	answered atomic.Bool

internal/agent/upgrade.go

@@ -35,12 +35,25 @@ func (a *Agent) upgrade(ctx context.Context, m *llmgwv1.UpgradeAgent, out chan<-
 		select {
 		case out <- &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Upgrade{Upgrade: reply}}:
 		case <-ctx.Done():
+		case <-time.After(5 * time.Second):
 		}
 	}
+	// One at a time: a second request while one downloads would race it to
+	// the same file.
+	if !a.upgrading.CompareAndSwap(false, true) {
+		reply.Error = "an upgrade is already under way on this node"
+		send()
+		return
+	}
+	defer a.upgrading.Store(false)
 	a.log.Info("upgrading", "to", m.GetVersion(), "from", a.cfg.Version)
 	self, err := upgradeTarget()
 	if err == nil {
-		dctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
+		// Not the stream's context: a stream that breaks while the release
+		// downloads — a control plane restarting, leadership moving — cancelled
+		// the upgrade with nobody told. The node registering again on the
+		// version is what the control plane watches for.
+		dctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), upgradeDownload)
 		_, err = release.Install(dctx, nil, m.GetReleases(), m.GetVersion(), runtime.GOOS, runtime.GOARCH, self)
 		cancel()
 	}
@@ -53,6 +66,13 @@ func (a *Agent) upgrade(ctx context.Context, m *llmgwv1.UpgradeAgent, out chan<-
 	send()
 	a.log.Info("upgraded; restarting into the new version", "to", m.GetVersion())
 	// A moment for the reply to leave before the stream goes.
-	time.Sleep(500 * time.Millisecond)
+	time.Sleep(upgradeLinger)
 	exitForUpgrade()
 }
+
+// upgradeDownload bounds an upgrade's download; the control plane waits
+// longer (agentsvc.UpgradeWait). upgradeLinger is the moment its answer has
+// to leave before the agent exits; a variable so tests need not wait it.
+const upgradeDownload = 10 * time.Minute
+
+var upgradeLinger = 2 * time.Second

internal/cli/ctl_upgrade.go

@@ -11,11 +11,13 @@ import (
 	"runtime"
 	"slices"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/spf13/cobra"
 
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/onboard"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/release"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/version"
@@ -244,6 +246,7 @@ func upgradeNodes(ctx context.Context, out io.Writer, client func() *ctlClient,
 	}
 	slices.SortFunc(nodes, func(a, b adminapi.NodeView) int { return strings.Compare(a.ID, b.ID) })
 	var todo []adminapi.NodeView
+	var away []string
 	skipped := 0
 	for _, n := range nodes {
 		if len(only) > 0 && !slices.Contains(only, n.ID) {
@@ -253,8 +256,17 @@ func upgradeNodes(ctx context.Context, out io.Writer, client func() *ctlClient,
 			skipped++
 			continue
 		}
+		// A node that is not connected cannot be told, and one tried first
+		// stopped every run at it: passed over, and named at the end.
+		if n.State == "lost" || n.State == "unreachable" {
+			away = append(away, n.ID)
+			continue
+		}
 		todo = append(todo, n)
 	}
+	if len(away) > 0 {
+		fmt.Fprintf(out, "Passed over, not connected: %s\n", strings.Join(away, ", "))
+	}
 	parallel = max(parallel, 1)
 	start, size, done := time.Now(), 1, 0
 	for len(todo) > 0 {
@@ -296,7 +308,10 @@ func upgradeNodes(ctx context.Context, out io.Writer, client func() *ctlClient,
 }
 
 // upgradeWave tells every node in the wave at once, and waits until each is
-// back on v, ready, with as many ready replicas as it had.
+// back on v and serving what it served.
+//
+// Told together, not one after another: each answer can take minutes, and a
+// wave of twenty waited for them in turn with nothing on the screen.
 func upgradeWave(ctx context.Context, out io.Writer, client func() *ctlClient, v string,
 	wave []adminapi.NodeView, perNode time.Duration,
 ) error {
@@ -308,35 +323,61 @@ func upgradeWave(ctx context.Context, out io.Writer, client func() *ctlClient, v
 	}
 	fmt.Fprintf(out, "== %d node(s): %s\n", len(wave), abbreviate(ids, 8))
 	body, _ := json.Marshal(adminapi.UpgradeRequest{Version: v})
-	var failed []string
+	var (
+		mu     sync.Mutex
+		failed []string
+		wg     sync.WaitGroup
+	)
 	for _, n := range wave {
-		// The control plane waits for the agent to fetch and check the
-		// release, up to five minutes; fifteen seconds gave up on a node
-		// that was upgrading.
-		if err := client().patient(6*time.Minute).post(ctx, "/api/v1/nodes/"+n.ID+"/upgrade", body, nil); err != nil {
-			failed = append(failed, fmt.Sprintf("%s: %v", n.ID, err))
-		}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			// The control plane waits for the agent's answer, or for it to
+			// come back on v, longer than the agent gives its download.
+			err := client().patient(agentsvc.UpgradeWait+time.Minute).post(ctx, "/api/v1/nodes/"+n.ID+"/upgrade", body, nil)
+			mu.Lock()
+			defer mu.Unlock()
+			if err != nil {
+				failed = append(failed, fmt.Sprintf("%s: %v", n.ID, err))
+				fmt.Fprintf(out, "   %s: not upgraded: %v\n", n.ID, err)
+				return
+			}
+			fmt.Fprintf(out, "   %s: on %s, coming back\n", n.ID, v)
+		}()
 	}
-	if len(failed) > 0 {
-		return fmt.Errorf("not told to upgrade:\n  %s\n`llmgw ctl logs NODE` says more", strings.Join(failed, "\n  "))
-	}
-	waiting := slices.Clone(ids)
+	wg.Wait()
+	waiting := slices.DeleteFunc(slices.Clone(ids), func(id string) bool {
+		return slices.ContainsFunc(failed, func(f string) bool { return strings.HasPrefix(f, id+": ") })
+	})
 	deadline := time.Now().Add(perNode)
-	for {
+	backAt := map[string]time.Time{}
+	said := time.Now()
+	for len(waiting) > 0 {
 		var now []adminapi.NodeView
 		_ = client().get(ctx, "/api/v1/nodes", &now)
-		after := readyReplicasByNode(ctx, client)
+		var reps []adminapi.ReplicaView
+		_ = client().get(ctx, "/api/v1/replicas", &reps)
+		after, starting := replicasByNode(reps)
 		state := map[string]adminapi.NodeView{}
 		for _, x := range now {
 			state[x.ID] = x
 		}
 		waiting = slices.DeleteFunc(waiting, func(id string) bool {
 			cur := state[id]
-			return cur.AgentVersion == v && cur.State == "ready" && after[id] >= before[id]
+			// Connected and on v: ready, or cordoned or draining as it
+			// was, which an upgrade does not undo.
+			if cur.AgentVersion != v || (cur.State != "ready" && cur.State != "cordoned" && cur.State != "draining") {
+				return false
+			}
+			if _, ok := backAt[id]; !ok {
+				backAt[id] = time.Now()
+			}
+			// Serving what it served; or, a while on, nothing on it still
+			// starting: a model that went idle meanwhile is not one lost.
+			return after[id] >= before[id] || (time.Since(backAt[id]) > 90*time.Second && starting[id] == 0)
 		})
 		if len(waiting) == 0 {
-			fmt.Fprintf(out, "   done in %s\n", time.Since(start).Round(time.Second))
-			return nil
+			break
 		}
 		if time.Now().After(deadline) {
 			var why []string
@@ -345,11 +386,49 @@ func upgradeWave(ctx context.Context, out io.Writer, client func() *ctlClient, v
 				why = append(why, fmt.Sprintf("%s (version %q, state %q, replicas %d/%d)",
 					id, cur.AgentVersion, cur.State, after[id], before[id]))
 			}
-			return fmt.Errorf("not back within %s:\n  %s\n`llmgw ctl logs NODE` says why, and each keeps its "+
-				"previous binary beside the new one", perNode, strings.Join(why, "\n  "))
+			failed = append(failed, fmt.Sprintf("not back within %s: %s", perNode, strings.Join(why, ", ")))
+			break
+		}
+		if time.Since(said) >= 30*time.Second {
+			var w []string
+			for _, id := range waiting {
+				cur := state[id]
+				w = append(w, fmt.Sprintf("%s (%s, %s, replicas %d/%d)", id, orUnknown(cur.AgentVersion), cur.State,
+					after[id], before[id]))
+			}
+			fmt.Fprintf(out, "   waiting for %s\n", abbreviate(w, 8))
+			said = time.Now()
 		}
 		time.Sleep(2 * time.Second)
 	}
+	if len(failed) > 0 {
+		return fmt.Errorf("not upgraded:\n  %s\n`llmgw ctl logs NODE` says why; each keeps its previous binary beside the new one",
+			strings.Join(failed, "\n  "))
+	}
+	fmt.Fprintf(out, "   done in %s\n", time.Since(start).Round(time.Second))
+	return nil
+}
+
+// replicasByNode counts each node's ready replicas, and those still placing
+// or starting.
+func replicasByNode(reps []adminapi.ReplicaView) (ready, starting map[string]int) {
+	ready, starting = map[string]int{}, map[string]int{}
+	for _, r := range reps {
+		seen := map[string]bool{}
+		for _, sh := range r.Shards {
+			if seen[sh.Node] {
+				continue
+			}
+			seen[sh.Node] = true
+			switch r.State {
+			case "ready":
+				ready[sh.Node]++
+			case "placing", "starting":
+				starting[sh.Node]++
+			}
+		}
+	}
+	return ready, starting
 }
 
 // abbreviate lists up to n names, and says how many more.

internal/cli/ctl_upgrade_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"slices"
 	"strings"
 	"sync"
 	"testing"
@@ -21,6 +22,8 @@ type fakeFleet struct {
 	versions map[string]string
 	stuck    map[string]bool
 	told     []string
+	// states are nodes' states other than ready.
+	states map[string]string
 }
 
 func (f *fakeFleet) server(t *testing.T) *httptest.Server {
@@ -31,7 +34,11 @@ func (f *fakeFleet) server(t *testing.T) *httptest.Server {
 		case r.URL.Path == "/api/v1/nodes":
 			var out []adminapi.NodeView
 			for id, v := range f.versions {
-				out = append(out, adminapi.NodeView{ID: id, State: "ready", AgentVersion: v})
+				st := "ready"
+				if f.states[id] != "" {
+					st = f.states[id]
+				}
+				out = append(out, adminapi.NodeView{ID: id, State: st, AgentVersion: v})
 			}
 			_ = json.NewEncoder(w).Encode(out)
 		case r.URL.Path == "/api/v1/replicas":
@@ -201,3 +208,23 @@ func TestUpgradingAControlPlaneAloneSaysItsNodesWereNot(t *testing.T) {
 		t.Errorf("said something with every node on the version: %s", out.String())
 	}
 }
+
+// A cordoned node upgraded is done, as it was; a node not connected is passed
+// over rather than stopping every node after it.
+func TestAnUpgradeTakesACordonedNodeAndPassesOverOneAway(t *testing.T) {
+	f := &fakeFleet{versions: map[string]string{"a": "v1.0.0", "b": "v1.0.0", "c": "v1.0.0"},
+		states: map[string]string{"a": "lost", "b": "cordoned"}}
+	srv := f.server(t)
+	var out bytes.Buffer
+	err := upgradeNodes(t.Context(), &out, func() *ctlClient { return newCtlClient(srv.URL) }, "v1.0.1", nil, 5*time.Second, 2)
+	if slices.Contains(f.told, "a") || !slices.Contains(f.told, "b") || !slices.Contains(f.told, "c") {
+		t.Errorf("told %v", f.told)
+	}
+	if !strings.Contains(out.String(), "Passed over, not connected: a") {
+		t.Errorf("output:\n%s", out.String())
+	}
+	// The one away is named at the end, behind.
+	if err == nil || !strings.Contains(err.Error(), "a (v1.0.0, lost)") {
+		t.Errorf("err %v", err)
+	}
+}

internal/cli/serverinit_cluster.go

@@ -64,6 +64,9 @@ const agentUnit = `[Unit]
 Description=llmgw node agent
 After=network-online.target
 Wants=network-online.target
+# Never given up on: a new release that fails its first starts, or an upgrade
+# that restarts it twice in a minute, would otherwise leave the node off.
+StartLimitIntervalSec=0
 
 [Service]
 ExecStart=/usr/local/bin/llmgw agent --config /etc/llmgw/agent.yaml

internal/cli/tui_actions.go

@@ -16,6 +16,7 @@ import (
 
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/facility"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
@@ -571,9 +572,15 @@ func (m consoleModel) nodeMenu(n adminapi.NodeView) *menuDialog {
 		menuItem{"Upgrade its agent…", func(m *consoleModel) tea.Cmd {
 			m.openForm(&formDialog{title: "Upgrade " + n.ID,
 				note:   "It fetches the release for its platform, checks it, swaps it in and restarts; engines keep serving.",
-				fields: []formField{textF("version", version.Version, "a release, like v0.11.1")},
+				fields: []formField{textF("version", releaseOrEmpty(version.Version), "a release, like v0.11.1")},
 				submit: func(m *consoleModel, v []string) (tea.Cmd, error) {
-					return m.call("Upgrade "+n.ID, http.MethodPost, path+"/upgrade", adminapi.UpgradeRequest{Version: v[0]}, nil), nil
+					if !release.Tag.MatchString(v[0]) {
+						return nil, fmt.Errorf("a release is vMAJOR.MINOR.PATCH, like v0.15.3")
+					}
+					// As long as the control plane waits: the download, the
+					// restart, the node back on the version.
+					return m.callWaiting("Upgrade "+n.ID, http.MethodPost, path+"/upgrade", adminapi.UpgradeRequest{Version: v[0]},
+						func(json.RawMessage) string { return n.ID + " is on " + v[0] + "." }, agentsvc.UpgradeWait+time.Minute), nil
 				}})
 			return nil
 		}},
@@ -1056,3 +1063,12 @@ func purgeSays(raw json.RawMessage) string {
 	}
 	return s
 }
+
+// releaseOrEmpty is this build's version when it is a release, else empty:
+// a development build's is not one a node can fetch.
+func releaseOrEmpty(v string) string {
+	if release.Tag.MatchString(v) {
+		return v
+	}
+	return ""
+}

internal/control/agentsvc/heartbeat_test.go

@@ -3,6 +3,7 @@ package agentsvc
 import (
 	"log/slog"
 	"testing"
+	"time"
 
 	llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
@@ -284,6 +285,28 @@ func TestAnUpgradeWaitsForTheAgentsAnswer(t *testing.T) {
 	}
 }
 
+// The agent's answer lost — it exits a moment after sending it — and the node
+// back on the version: the upgrade is done, not five minutes of waiting and a
+// failure blaming an old agent.
+func TestAnUpgradeIsDoneWhenTheNodeComesBackOnTheVersion(t *testing.T) {
+	s := newTestService()
+	nodeWith(s, inventory.Node{ID: "n1", State: inventory.Ready, AgentVersion: "v1.2.2"})
+	c := &conn{nodeID: "n1", send: make(chan *llmgwv1.ControlMessage, 4), closed: make(chan struct{})}
+	s.conns["n1"] = c
+	done := make(chan error, 1)
+	go func() { done <- s.Upgrade(t.Context(), "n1", "v1.2.3", "https://r") }()
+	<-c.send
+	nodeWith(s, inventory.Node{ID: "n1", State: inventory.Ready, AgentVersion: "v1.2.3"})
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("still waiting for an answer the node will never send")
+	}
+}
+
 func TestAnotherTenantOnACardIsJudgedEveryHeartbeat(t *testing.T) {
 	// Found on cuda01: ComfyUI held 5 GB when the agent started, so the card
 	// was kept out of placement. ComfyUI let go of it and the card stayed

internal/control/agentsvc/service.go

@@ -1158,21 +1158,45 @@ func (s *Service) Upgrade(ctx context.Context, nodeID, version, releases string)
 		return err
 	}
 	s.log.Info("told a node to upgrade", "node", nodeID, "version", version)
-	// A download of some tens of megabytes on a slow link, and no longer: an
-	// agent too old to know this message never answers.
-	ctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
+	// Longer than the agent gives its download, so its own answer — done, or
+	// why not — is what is heard. Its answer can still be lost: it exits a
+	// moment after sending it, and a stream that breaks takes it along. The
+	// node registering again on the version says the same, and is watched for.
+	ctx, cancel := context.WithTimeout(ctx, UpgradeWait)
 	defer cancel()
-	select {
-	case r := <-ch:
-		if r.GetError() != "" {
-			return errors.New(r.GetError())
+	tick := time.NewTicker(2 * time.Second)
+	defer tick.Stop()
+	for {
+		select {
+		case r := <-ch:
+			if r.GetError() != "" {
+				return errors.New(r.GetError())
+			}
+			return nil
+		case <-tick.C:
+			if n, ok := s.inv.Get(nodeID); ok && n.AgentVersion == version {
+				return nil
+			}
+		case <-ctx.Done():
+			n, ok := s.inv.Get(nodeID)
+			switch {
+			case !slices.Contains(s.ConnectedNodes(), nodeID):
+				return fmt.Errorf("node %s went away during the upgrade and has not come back: on it, "+
+					"`systemctl status llmgw-agent` and `journalctl -u llmgw-agent` say why", nodeID)
+			case ok && n.AgentVersion != "":
+				return fmt.Errorf("node %s did not answer the upgrade and still runs %s: its download is slow or stuck, "+
+					"or its agent is older than v0.10.4, which cannot upgrade itself", nodeID, n.AgentVersion)
+			default:
+				return fmt.Errorf("node %s did not answer the upgrade; an agent older than v0.10.4 cannot upgrade itself", nodeID)
+			}
 		}
-		return nil
-	case <-ctx.Done():
-		return fmt.Errorf("node %s did not answer the upgrade; an agent older than v0.10.4 cannot upgrade itself", nodeID)
 	}
 }
 
+// UpgradeWait is how long the control plane waits for an agent to upgrade:
+// longer than the agent gives its download (agent.upgradeDownload).
+const UpgradeWait = 11 * time.Minute
+
 // wearFrom is a card's wear, when the agent read it.
 func wearFrom(g *llmgwv1.GPUSensor) *inventory.GPUWear {
 	if !g.GetHasWear() {