diff --git a/core/services/nodes/authenticated_routes_test.go b/core/services/nodes/authenticated_routes_test.go index db6c2a72b..215e2bead 100644 --- a/core/services/nodes/authenticated_routes_test.go +++ b/core/services/nodes/authenticated_routes_test.go @@ -72,6 +72,26 @@ var _ = Describe("extra authenticated routes on the worker HTTP server", func() Expect(get("/healthz", "").StatusCode).To(Equal(http.StatusOK)) }) + It("refuses to start when a route set names no prefix", func() { + lis, err := net.Listen("tcp", "127.0.0.1:0") + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = lis.Close() }) + dir := GinkgoT().TempDir() + _, err = StartFileTransferServerWithRoutes(lis, dir, dir, dir, token, 0, nil, + &AuthenticatedRoutes{Register: func(*http.ServeMux) {}}) + Expect(err).To(MatchError(ContainSubstring("prefix"))) + }) + + It("refuses to start when a route set names no registrar", func() { + lis, err := net.Listen("tcp", "127.0.0.1:0") + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = lis.Close() }) + dir := GinkgoT().TempDir() + _, err = StartFileTransferServerWithRoutes(lis, dir, dir, dir, token, 0, nil, + &AuthenticatedRoutes{Prefix: "/v1/control/"}) + Expect(err).To(MatchError(ContainSubstring("registrar"))) + }) + It("mounts nothing when no route set is given", func() { lis, err := net.Listen("tcp", "127.0.0.1:0") Expect(err).NotTo(HaveOccurred()) diff --git a/core/services/nodes/file_transfer_server.go b/core/services/nodes/file_transfer_server.go index 8232d9f64..3f3a4ebcc 100644 --- a/core/services/nodes/file_transfer_server.go +++ b/core/services/nodes/file_transfer_server.go @@ -90,6 +90,19 @@ func StartFileTransferServerWithReadiness(lis net.Listener, stagingDir, modelsDi // StartFileTransferServerWithRoutes is StartFileTransferServerWithReadiness // plus an extra authenticated route set. See AuthenticatedRoutes. func StartFileTransferServerWithRoutes(lis net.Listener, stagingDir, modelsDir, dataDir, token string, maxUploadSize int64, readiness *WorkerReadiness, extra *AuthenticatedRoutes, logStore ...*model.BackendLogStore) (*http.Server, error) { + // Checked before anything is created. A route set that names no prefix or + // no registrar is a caller bug, and mounting nothing for it would be the + // worst possible answer: the server comes up healthy and every route the + // caller believes it registered answers 404, which through a tunnel is + // indistinguishable from a version skew. + if extra != nil { + if extra.Prefix == "" { + return nil, fmt.Errorf("extra routes were given no prefix to mount under") + } + if extra.Register == nil { + return nil, fmt.Errorf("extra routes under %q were given no registrar", extra.Prefix) + } + } if err := os.MkdirAll(stagingDir, 0750); err != nil { return nil, fmt.Errorf("creating staging dir %s: %w", stagingDir, err) } @@ -189,7 +202,7 @@ func StartFileTransferServerWithRoutes(lis net.Listener, stagingDir, modelsDir, // Readiness: "can this worker actually accept work?" See WorkerReadiness. mux.HandleFunc("/readyz", probe(readiness.Check)) - if extra != nil && extra.Register != nil && extra.Prefix != "" { + if extra != nil { extraMux := http.NewServeMux() extra.Register(extraMux) mux.Handle(extra.Prefix, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/core/services/nodes/install_progress_publisher.go b/core/services/nodes/install_progress_publisher.go index 88a2d16ee..23bd67195 100644 --- a/core/services/nodes/install_progress_publisher.go +++ b/core/services/nodes/install_progress_publisher.go @@ -11,11 +11,10 @@ import ( // and hands them to its emit sink at most once per `interval`. Always emits the // final event on Flush so the UI sees the terminal percentage. // -// The sink is a function rather than a NATS subject because the same debounce -// now bounds two carriers: the NATS progress subject an agent node still -// publishes on, and a line written into the streaming HTTP response a worker -// serves over its tunnel. Keeping one debouncer keeps the ~4/s tick bound a -// single fact instead of one per carrier. +// The sink is a function rather than a NATS subject because the events now ride +// the streaming HTTP response the worker serves over its tunnel. Leaving the +// debounce here, rather than at the sink, is what keeps the ~4/s tick bound a +// property of install progress itself and not of whatever carries it. // // Behavior: leading-edge debounce. The first OnDownload after a quiet window // publishes immediately; subsequent ticks within `interval` only buffer the @@ -38,21 +37,17 @@ type DebouncedInstallProgressPublisher struct { timer *time.Timer } -// NewDebouncedInstallProgressPublisher constructs a publisher for one install -// operation that publishes to the per-op NATS progress subject. interval is the -// leading-edge debounce window (~250ms in production). -func NewDebouncedInstallProgressPublisher(client messaging.MessagingClient, nodeID, opID, backend string, interval time.Duration) *DebouncedInstallProgressPublisher { - subject := messaging.SubjectNodeBackendInstallProgress(nodeID, opID) - return NewDebouncedInstallProgressSink(func(ev messaging.BackendInstallProgressEvent) { - _ = client.Publish(subject, ev) - }, nodeID, opID, backend, interval) -} - // NewDebouncedInstallProgressSink constructs a publisher for one install // operation that hands each debounced event to emit. // // emit is called with p.mu released, so a sink that blocks on a slow link // cannot stall the gallery download loop that feeds it. +// +// There was a sibling constructor that published to the per-op NATS progress +// subject. It is gone rather than kept: a worker streams these events inside +// the install response now, so the NATS publisher had no caller left, and a +// constructor alive only for its own spec is a carrier a reader would believe +// still runs. func NewDebouncedInstallProgressSink(emit func(messaging.BackendInstallProgressEvent), nodeID, opID, backend string, interval time.Duration) *DebouncedInstallProgressPublisher { return &DebouncedInstallProgressPublisher{ emit: emit, diff --git a/core/services/nodes/install_progress_publisher_test.go b/core/services/nodes/install_progress_publisher_test.go index 04073cebe..a0c42c9fd 100644 --- a/core/services/nodes/install_progress_publisher_test.go +++ b/core/services/nodes/install_progress_publisher_test.go @@ -1,6 +1,7 @@ package nodes import ( + "sync" "time" . "github.com/onsi/ginkgo/v2" @@ -9,10 +10,29 @@ import ( "github.com/mudler/LocalAI/core/services/messaging" ) +// collectingSink records every event the debouncer emits. Emits arrive from the +// trailing timer's own goroutine as well as from OnDownload, so it locks. +type collectingSink struct { + mu sync.Mutex + events []messaging.BackendInstallProgressEvent +} + +func (c *collectingSink) emit(ev messaging.BackendInstallProgressEvent) { + c.mu.Lock() + defer c.mu.Unlock() + c.events = append(c.events, ev) +} + +func (c *collectingSink) snapshot() []messaging.BackendInstallProgressEvent { + c.mu.Lock() + defer c.mu.Unlock() + return append([]messaging.BackendInstallProgressEvent(nil), c.events...) +} + var _ = Describe("DebouncedInstallProgressPublisher", func() { - It("publishes the first event immediately and debounces subsequent ones within the window", func() { - mc := newScriptedMessagingClient() - pub := NewDebouncedInstallProgressPublisher(mc, "n1", "op1", "vllm", 50*time.Millisecond) + It("emits the first event immediately and debounces subsequent ones within the window", func() { + sink := &collectingSink{} + pub := NewDebouncedInstallProgressSink(sink.emit, "n1", "op1", "vllm", 50*time.Millisecond) // Three rapid-fire ticks within the debounce window. pub.OnDownload("vllm.tar.zst", "100 MB", "1 GB", 10.0) @@ -20,29 +40,44 @@ var _ = Describe("DebouncedInstallProgressPublisher", func() { pub.OnDownload("vllm.tar.zst", "300 MB", "1 GB", 30.0) pub.Flush() - // First event publishes immediately; the others coalesce; Flush guarantees a final. - // So we expect at least 2 publishes and at most 4 (lead + final + any window-bounded). - Eventually(func() int { - return len(mc.publishCalls(messaging.SubjectNodeBackendInstallProgress("n1", "op1"))) - }, "1s").Should(BeNumerically(">=", 2)) - calls := mc.publishCalls(messaging.SubjectNodeBackendInstallProgress("n1", "op1")) - Expect(len(calls)).To(BeNumerically("<=", 4), - "three ticks within the debounce window should produce at most ~4 publishes") + // First event emits immediately; the others coalesce; Flush guarantees a final. + // So we expect at least 2 emits and at most 4 (lead + final + any window-bounded). + Eventually(func() int { return len(sink.snapshot()) }, "1s").Should(BeNumerically(">=", 2)) + Expect(len(sink.snapshot())).To(BeNumerically("<=", 4), + "three ticks within the debounce window should produce at most ~4 emits") }) - It("publishes the final event after Flush with the latest percentage", func() { - mc := newScriptedMessagingClient() - pub := NewDebouncedInstallProgressPublisher(mc, "n1", "op1", "vllm", 50*time.Millisecond) + It("emits the final event after Flush with the latest percentage", func() { + sink := &collectingSink{} + pub := NewDebouncedInstallProgressSink(sink.emit, "n1", "op1", "vllm", 50*time.Millisecond) pub.OnDownload("vllm.tar.zst", "1 GB", "1 GB", 100.0) pub.Flush() Eventually(func() float64 { - calls := mc.publishCalls(messaging.SubjectNodeBackendInstallProgress("n1", "op1")) - if len(calls) == 0 { + events := sink.snapshot() + if len(events) == 0 { return -1 } - return calls[len(calls)-1].Percentage + return events[len(events)-1].Percentage }, "1s").Should(Equal(100.0)) }) + + It("stamps every event with the identity the frontend correlates on", func() { + // The op id and node id are what a frontend matches a progress line to + // an operation with; the subject used to carry them and now nothing + // else does, so the event body has to. + sink := &collectingSink{} + pub := NewDebouncedInstallProgressSink(sink.emit, "node-7", "op-42", "vllm", time.Millisecond) + pub.OnDownload("vllm.tar.zst", "1 GB", "1 GB", 100.0) + pub.Flush() + + Eventually(func() int { return len(sink.snapshot()) }, "1s").Should(BeNumerically(">=", 1)) + ev := sink.snapshot()[0] + Expect(ev.NodeID).To(Equal("node-7")) + Expect(ev.OpID).To(Equal("op-42")) + Expect(ev.Backend).To(Equal("vllm")) + Expect(ev.FileName).To(Equal("vllm.tar.zst")) + Expect(ev.Phase).To(Equal(messaging.PhaseDownloading)) + }) }) diff --git a/core/services/worker/control_routes.go b/core/services/worker/control_routes.go index dec8e70fc..d787f23cc 100644 --- a/core/services/worker/control_routes.go +++ b/core/services/worker/control_routes.go @@ -7,6 +7,7 @@ import ( "io" "net/http" "sync" + "unicode/utf8" "github.com/mudler/LocalAI/core/services/messaging" "github.com/mudler/LocalAI/core/services/workerctl" @@ -117,6 +118,14 @@ func (s *backendSupervisor) RegisterControlRoutes(mux *http.ServeMux) { return messaging.ModelsRunningReply{Models: s.runningModels()}, nil }) + // model.stop deliberately does NOT take the caller's context, and the + // asymmetry with model.unload below is the point. This is the ACKNOWLEDGED + // stop path: it reserves the process, frees it, kills it, waits for it to + // exit and releases its port. Abandoning that half way because the caller + // stopped listening would leave a process the worker has marked stopping, + // a port not returned to the allocator, and a controller row nothing ever + // reconciles. The stop has to finish whether or not anyone reads the + // answer; its own bounds are internal and already in place. post(workerctl.PathModelStop, func(_ context.Context, body []byte) (any, error) { var req messaging.ModelStopRequest if err := json.Unmarshal(body, &req); err != nil { @@ -137,6 +146,10 @@ func (s *backendSupervisor) RegisterControlRoutes(mux *http.ServeMux) { return s.deleteBackend(req), nil }) + // model.unload is the one verb that DOES take the caller's context. Free is + // the whole operation here rather than a courtesy before a kill: nothing is + // terminated, no port is released, and abandoning it leaves the worker + // exactly as it was. So the caller's budget is the operation's budget. post(workerctl.PathModelUnload, func(ctx context.Context, body []byte) (any, error) { var req messaging.ModelUnloadRequest if err := json.Unmarshal(body, &req); err != nil { @@ -153,6 +166,9 @@ func (s *backendSupervisor) RegisterControlRoutes(mux *http.ServeMux) { return s.deleteModel(req), nil }) + // backend.stop drops the caller's context for the same reason model.stop + // does, and more plainly: it is fire-and-forget, so there is no answer for + // the caller to still be waiting on. post(workerctl.PathBackendStop, func(_ context.Context, body []byte) (any, error) { req, stopAll, err := decodeBackendStopRequest(body) if err != nil { @@ -207,24 +223,24 @@ func readControlBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) { } // truncate bounds a caller-controlled string that is about to be echoed. -// It cuts on a rune boundary: a byte-wise cut can split a multi-byte rune, and -// the half-rune then travels as a replacement character through every log and -// UI that reads it. +// +// It cuts on a rune boundary. A byte-wise cut can split a multi-byte rune, and +// the half rune then travels as a replacement character through every log and +// UI that reads it; phase 2 shipped exactly that defect on a refusal reason and +// pinned the rule afterwards. utf8.RuneStart is the same predicate the cluster +// package uses for it, so the two are one rule rather than two hand-rolled +// copies that can drift. func truncate(s string, max int) string { if len(s) <= max { return s } cut := max - for cut > 0 && !isRuneStart(s[cut]) { + for cut > 0 && !utf8.RuneStart(s[cut]) { cut-- } return s[:cut] + "…" } -// isRuneStart reports whether b begins a UTF-8 rune (i.e. is not a -// continuation byte). -func isRuneStart(b byte) bool { return b&0xC0 != 0x80 } - // ndjsonStream writes the Envelope lines of one streaming control response. // // The mutex is not optional. A progress line can be emitted from the debounce diff --git a/core/services/worker/control_routes_test.go b/core/services/worker/control_routes_test.go index fb8576e71..f57793324 100644 --- a/core/services/worker/control_routes_test.go +++ b/core/services/worker/control_routes_test.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "io" + "net" "net/http" "net/http/httptest" "os" @@ -13,6 +14,7 @@ import ( "strings" "sync" "syscall" + "unicode/utf8" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -20,6 +22,7 @@ import ( "github.com/mudler/LocalAI/core/services/messaging" "github.com/mudler/LocalAI/core/services/nodes" "github.com/mudler/LocalAI/core/services/workerctl" + "github.com/mudler/LocalAI/pkg/model" "github.com/mudler/LocalAI/pkg/system" ) @@ -136,6 +139,39 @@ var _ = Describe("worker control routes", func() { Expect(string(body)).To(ContainSubstring("control")) }) + It("cuts the echoed path on a rune boundary, so no half rune reaches a log", func() { + // The multi-byte rune is placed so that it STRADDLES the cut: a + // byte-wise cut lands inside it and the body carries a replacement + // character. Phase 2 shipped this exact defect on a refusal reason. + // "a" is one byte and "€" is three. The echoed string is the WHOLE + // path, prefix included, so the padding is sized against the prefix to + // put the rune across bytes max-1, max and max+1. + lead := maxEchoedPathBytes - len(workerctl.Prefix) - 1 + straddle := strings.Repeat("a", lead) + "€" + strings.Repeat("b", 64) + resp := post(workerctl.Prefix+straddle, struct{}{}) + Expect(resp.StatusCode).To(Equal(http.StatusNotFound)) + body, err := io.ReadAll(resp.Body) + Expect(err).NotTo(HaveOccurred()) + + Expect(utf8.ValidString(string(body))).To(BeTrue(), "the 404 body must stay valid UTF-8") + Expect(string(body)).NotTo(ContainSubstring("\uFFFD"), "the cut split a rune") + // And the whole rune is dropped rather than kept past the bound. + Expect(string(body)).NotTo(ContainSubstring("€")) + }) + + It("keeps a rune that ends exactly on the bound, so the cut is not off by one", func() { + // Here the rune's last byte is at maxEchoedPathBytes-1, so it fits + // entirely and must survive. A cut that walked back unconditionally + // would drop it and this spec would catch that. + lead := maxEchoedPathBytes - len(workerctl.Prefix) - 3 + exact := strings.Repeat("a", lead) + "€" + strings.Repeat("b", 64) + resp := post(workerctl.Prefix+exact, struct{}{}) + body, err := io.ReadAll(resp.Body) + Expect(err).NotTo(HaveOccurred()) + Expect(utf8.ValidString(string(body))).To(BeTrue()) + Expect(string(body)).To(ContainSubstring("€")) + }) + It("bounds the unknown path it echoes back, so a long URL cannot be reflected wholesale", func() { long := strings.Repeat("a", 4096) resp := post(workerctl.Prefix+long, struct{}{}) @@ -196,6 +232,8 @@ var _ = Describe("worker control routes", func() { }) It("answers model.unload with the worker's own reply rather than a transport error", func() { + // Nothing is loaded, so there is nothing to free and the true answer is + // success. resp := post(workerctl.PathModelUnload, messaging.ModelUnloadRequest{ModelName: "m"}) Expect(resp.StatusCode).To(Equal(http.StatusOK)) var reply messaging.ModelUnloadReply @@ -203,6 +241,26 @@ var _ = Describe("worker control routes", func() { Expect(reply.Success).To(BeTrue()) }) + It("reports a failed Free as a failed unload, not as success", func() { + // A worker that answers "done" about work it did not do is the fourth + // kind of answer in this programme's taxonomy, and it is the one acted + // on: the frontend's only caller of unload is EvictLRU, so a false yes + // tells the scheduler VRAM was released and lets it place the next + // model on a node still holding the old one. + dead, err := net.Listen("tcp", "127.0.0.1:0") + Expect(err).NotTo(HaveOccurred()) + addr := dead.Addr().String() + Expect(dead.Close()).To(Succeed()) + + resp := post(workerctl.PathModelUnload, messaging.ModelUnloadRequest{ModelName: "m", Address: addr}) + // Still a 200: the WORKER answered. Only the verdict inside is negative. + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + var reply messaging.ModelUnloadReply + Expect(json.NewDecoder(resp.Body).Decode(&reply)).To(Succeed()) + Expect(reply.Success).To(BeFalse()) + Expect(reply.Error).To(ContainSubstring(addr)) + }) + It("answers model.stop for an unknown process with the worker's verdict, not a 5xx", func() { // The whole invariant of the phase: "that backend is not there" is the // WORKER answering, and it must never arrive as the status code a @@ -522,3 +580,137 @@ var _ = Describe("the NDJSON control stream", func() { } }) }) + +// These specs drive the REAL installBackend and upgradeBackend, with no +// installFn override, so the progress wiring inside them is exercised end to +// end through the carrier. Every other install spec scripts installFn, which +// leaves the one line that decides whether a caller sees any progress at all +// pinned by nothing. +var _ = Describe("the real install progress wiring", func() { + var ( + sup *backendSupervisor + srv *httptest.Server + ) + + BeforeEach(func() { + dir := GinkgoT().TempDir() + st, err := system.GetSystemState( + system.WithModelPath(dir), + system.WithBackendPath(filepath.Join(dir, "backends")), + system.WithBackendSystemPath(filepath.Join(dir, "backends-system")), + ) + Expect(err).NotTo(HaveOccurred()) + sup = &backendSupervisor{ + cfg: &Config{ModelsPath: dir, BackendsPath: filepath.Join(dir, "backends")}, + systemState: st, + ml: model.NewModelLoader(st), + nodeID: "node-under-test", + sigCh: make(chan os.Signal, 1), + processes: map[string]*backendProcess{}, + // Empty on purpose. The install cannot succeed, which is the point: + // what is under test is that the caller is told what the worker is + // doing and then told it failed, not that a download works. + galleries: nil, + } + mux := http.NewServeMux() + sup.RegisterControlRoutes(mux) + srv = httptest.NewServer(mux) + DeferCleanup(srv.Close) + }) + + postJSON := func(path string, body any) *http.Response { + GinkgoHelper() + buf, err := json.Marshal(body) + Expect(err).NotTo(HaveOccurred()) + resp, err := srv.Client().Post(srv.URL+path, "application/json", bytes.NewReader(buf)) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + return resp + } + + // envelopesOf decodes a whole NDJSON body into its envelopes. + envelopesOf := func(body io.Reader) []workerctl.Envelope { + GinkgoHelper() + var out []workerctl.Envelope + dec := json.NewDecoder(body) + for { + var env workerctl.Envelope + err := dec.Decode(&env) + if errors.Is(err, io.EOF) { + break + } + Expect(err).NotTo(HaveOccurred()) + out = append(out, env) + } + return out + } + + It("streams a resolving line before the failure, from the real installBackend", func() { + resp := postJSON(workerctl.PathBackendInstall, + messaging.BackendInstallRequest{Backend: "no-such-backend", OpID: "op-real"}) + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + + envs := envelopesOf(resp.Body) + Expect(envs).NotTo(BeEmpty()) + Expect(envs[0].Progress).NotTo(BeNil(), "the caller must see a line before the gallery work starts") + + var ev messaging.BackendInstallProgressEvent + Expect(json.Unmarshal(envs[0].Progress, &ev)).To(Succeed()) + Expect(ev.Phase).To(Equal(messaging.PhaseResolving)) + Expect(ev.OpID).To(Equal("op-real")) + Expect(ev.NodeID).To(Equal("node-under-test")) + Expect(ev.Backend).To(Equal("no-such-backend")) + + last := envs[len(envs)-1] + Expect(last.Reply).NotTo(BeNil()) + var reply messaging.BackendInstallReply + Expect(json.Unmarshal(last.Reply, &reply)).To(Succeed()) + Expect(reply.Success).To(BeFalse()) + }) + + It("streams nothing but the reply when the caller asked for no progress", func() { + // An empty OpID is a reconciler-driven retry. It must not be given a + // progress stream, and it must still get its terminal line. + resp := postJSON(workerctl.PathBackendInstall, + messaging.BackendInstallRequest{Backend: "no-such-backend"}) + Expect(envelopeKindsOf(resp.Body)).To(Equal([]string{"reply"})) + }) + + It("streams a resolving line from the real upgradeBackend too", func() { + resp := postJSON(workerctl.PathBackendUpgrade, + messaging.BackendUpgradeRequest{Backend: "no-such-backend", OpID: "op-real-upgrade"}) + envs := envelopesOf(resp.Body) + Expect(envs).NotTo(BeEmpty()) + Expect(envs[0].Progress).NotTo(BeNil()) + + var ev messaging.BackendInstallProgressEvent + Expect(json.Unmarshal(envs[0].Progress, &ev)).To(Succeed()) + Expect(ev.Phase).To(Equal(messaging.PhaseResolving)) + Expect(ev.OpID).To(Equal("op-real-upgrade")) + + var reply messaging.BackendUpgradeReply + Expect(json.Unmarshal(envs[len(envs)-1].Reply, &reply)).To(Succeed()) + Expect(reply.Success).To(BeFalse()) + }) + + It("hands the gallery no download callback when either half of the guard is missing", func() { + // A nil callback is what puts the gallery on its silent path, and both + // halves have to be checked: a caller with an OpID but no sink is the + // shape that would nil-panic on the resolving emit. + collected := func(messaging.BackendInstallProgressEvent) {} + + cb, flush := sup.startProgress("", "b", collected) + Expect(cb).To(BeNil()) + Expect(flush).NotTo(BeNil()) + flush() + + cb, flush = sup.startProgress("op", "b", nil) + Expect(cb).To(BeNil()) + Expect(flush).NotTo(BeNil()) + flush() + + cb, flush = sup.startProgress("op", "b", collected) + Expect(cb).NotTo(BeNil()) + flush() + }) +}) diff --git a/core/services/worker/install.go b/core/services/worker/install.go index 2390ca28d..1fc3238f4 100644 --- a/core/services/worker/install.go +++ b/core/services/worker/install.go @@ -134,19 +134,11 @@ func (s *backendSupervisor) installBackend(ctx context.Context, req messaging.Ba galleries = reqGalleries } - // When the caller tagged this install with an OpID and is listening for - // progress, stream the gallery download ticks back on the response the - // caller is already reading. Callers that omit OpID stay on the silent path. - // The publisher releases its mutex before every emit so a slow link never - // stalls the download loop, and the deferred Flush guarantees a - // terminal-percentage event reaches the caller even when the install errors - // out. - var downloadCb func(file, current, total string, percentage float64) - if req.OpID != "" && onProgress != nil { - publisher := nodes.NewDebouncedInstallProgressSink(onProgress, s.nodeID, req.OpID, req.Backend, installProgressDebounce) - downloadCb = publisher.OnDownload - defer publisher.Flush() - } + // Gallery download ticks go back on the response the caller is already + // reading. See startProgress for the guard and for why the flush is + // deferred. + downloadCb, flushProgress := s.startProgress(req.OpID, req.Backend, onProgress) + defer flushProgress() // On upgrade, run the gallery install path even if the binary already // exists on disk: findBackend would otherwise short-circuit and we'd @@ -235,17 +227,10 @@ func (s *backendSupervisor) upgradeBackend(ctx context.Context, req messaging.Ba galleries = reqGalleries } - // When the caller tagged this upgrade with an OpID, stream gallery download - // progress back on the same sink install uses — an upgrade IS a - // force-reinstall. Callers that omit OpID stay on the silent path. The - // deferred Flush guarantees a terminal-percentage event even if the upgrade - // errors out, so the caller's per-node bar never hangs mid-download. - var downloadCb func(file, current, total string, percentage float64) - if req.OpID != "" && onProgress != nil { - publisher := nodes.NewDebouncedInstallProgressSink(onProgress, s.nodeID, req.OpID, req.Backend, installProgressDebounce) - downloadCb = publisher.OnDownload - defer publisher.Flush() - } + // The same sink install uses: an upgrade IS a force-reinstall, so its + // progress is install progress. + downloadCb, flushProgress := s.startProgress(req.OpID, req.Backend, onProgress) + defer flushProgress() if req.URI != "" { xlog.Info("Upgrading backend from external URI", "backend", req.Backend, "uri", req.URI) @@ -309,3 +294,34 @@ func (s *backendSupervisor) lockBackend(name string) func() { m.Lock() return m.Unlock } + +// startProgress wires one install or upgrade to its caller's progress sink. +// +// It returns the download callback the gallery takes, and a flush the caller +// must defer: the debouncer buffers within its window, and the deferred flush +// is what gets the terminal percentage to the caller even when the install +// errors out. +// +// Both halves of the guard matter. An empty OpID means the caller is a +// reconciler-driven retry that asked for no progress, and a nil sink means the +// caller is not reading a stream at all; either way the gallery gets a nil +// callback and takes its silent path. +// +// The resolving event is emitted here, before any gallery work, so the caller +// sees a line as soon as the worker has accepted the job rather than only once +// bytes start moving. A cold install spends minutes resolving a manifest, and +// during that time a progress stream with nothing on it is indistinguishable +// from one that is broken. +func (s *backendSupervisor) startProgress(opID, backend string, onProgress func(messaging.BackendInstallProgressEvent)) (func(file, current, total string, percentage float64), func()) { + if opID == "" || onProgress == nil { + return nil, func() {} + } + publisher := nodes.NewDebouncedInstallProgressSink(onProgress, s.nodeID, opID, backend, installProgressDebounce) + onProgress(messaging.BackendInstallProgressEvent{ + OpID: opID, + NodeID: s.nodeID, + Backend: backend, + Phase: messaging.PhaseResolving, + }) + return publisher.OnDownload, publisher.Flush +} diff --git a/core/services/worker/lifecycle.go b/core/services/worker/lifecycle.go index 824b569e0..888a1f4e1 100644 --- a/core/services/worker/lifecycle.go +++ b/core/services/worker/lifecycle.go @@ -183,18 +183,34 @@ func (s *backendSupervisor) unloadModel(ctx context.Context, req messaging.Model s.mu.Unlock() } - if targetAddr != "" { - // Best-effort bounded gRPC Free(). A model.unload request must not - // occupy the handler forever when a backend is wedged. The bound is - // derived from the caller's own budget where it has one, so a caller - // that allowed less than workerBackendFreeTimeout is not made to wait - // longer than it asked for. - client := grpc.NewClientWithToken(targetAddr, false, nil, false, s.cfg.RegistrationToken) - freeCtx, cancel := context.WithTimeout(ctx, workerBackendFreeTimeout) - if err := client.Free(freeCtx); err != nil { - xlog.Warn("Free() failed during model.unload", "error", err, "addr", targetAddr) + if targetAddr == "" { + // Nothing is loaded here, so there is nothing to free. That is the + // worker's own answer and it is a true one, not a claim about work it + // performed. + return messaging.ModelUnloadReply{Success: true} + } + + // Bounded gRPC Free(). A model.unload request must not occupy the handler + // forever when a backend is wedged. The bound is derived from the caller's + // own budget, so a caller that allowed less than workerBackendFreeTimeout + // is not made to wait longer than it asked for. + client := grpc.NewClientWithToken(targetAddr, false, nil, false, s.cfg.RegistrationToken) + freeCtx, cancel := context.WithTimeout(ctx, workerBackendFreeTimeout) + err := client.Free(freeCtx) + cancel() + if err != nil { + // Reported, not swallowed. This used to answer Success:true whatever + // Free did, which is the worker saying "done" about something it did + // not do: the frontend's only caller is EvictLRU, so a false yes told + // the scheduler VRAM had been released and let it place the next model + // on a node still holding the old one. A failure here is the worker's + // own verdict about one Free call and nothing more; it is not a + // statement that the node or the model is gone. + xlog.Warn("Free() failed during model.unload", "error", err, "addr", targetAddr) + return messaging.ModelUnloadReply{ + Success: false, + Error: fmt.Sprintf("freeing model on %s: %v", targetAddr, err), } - cancel() } return messaging.ModelUnloadReply{Success: true} @@ -215,7 +231,7 @@ func (s *backendSupervisor) deleteModel(req messaging.ModelDeleteRequest) messag // signalNodeStop serves node.stop: it triggers the normal shutdown path via // sigCh so deferred cleanup runs, rather than exiting the process here. func (s *backendSupervisor) signalNodeStop() { - xlog.Info("Serving node.stop — signaling shutdown") + xlog.Info("Serving node.stop, signaling shutdown") select { case s.sigCh <- syscall.SIGTERM: default: diff --git a/core/services/worker/supervisor.go b/core/services/worker/supervisor.go index 3f90830f3..d06b9d69f 100644 --- a/core/services/worker/supervisor.go +++ b/core/services/worker/supervisor.go @@ -849,6 +849,12 @@ func (s *backendSupervisor) stopBackendExact(key string, force bool) error { } if !force { + // Background and not a caller's context on purpose. This Free is a + // courtesy before the process is killed anyway, and the stop that + // follows must complete for the port to be released, so binding it to + // a caller that may have gone would buy nothing and could abandon a + // half-finished stop. model.unload, where Free IS the operation, takes + // the caller's budget instead; see RegisterControlRoutes. client := grpc.NewClientWithToken(bp.addr, false, nil, false, s.cfg.RegistrationToken) freeCtx, cancel := context.WithTimeout(context.Background(), workerBackendFreeTimeout) xlog.Debug("Calling bounded Free() before stopping backend", "backend", key, "timeout", workerBackendFreeTimeout) @@ -895,6 +901,8 @@ func (s *backendSupervisor) stopModelExact(req messaging.ModelStopRequest) messa s.mu.Unlock() if !req.Force { + // Background, for the reason given in stopBackendExact: this is the + // acknowledged stop and it has to run to completion. client := grpc.NewClientWithToken(bp.addr, false, nil, false, s.cfg.RegistrationToken) freeCtx, cancel := context.WithTimeout(context.Background(), workerBackendFreeTimeout) freeErr := client.Free(freeCtx) diff --git a/core/services/workerctl/paths_test.go b/core/services/workerctl/paths_test.go index c5685a99d..8ffabb2e3 100644 --- a/core/services/workerctl/paths_test.go +++ b/core/services/workerctl/paths_test.go @@ -40,10 +40,24 @@ var _ = Describe("control plane paths on the wire", func() { } }) - It("enumerates every verb, so a new one cannot be added without the prefix check seeing it", func() { - Expect(workerctl.AllPaths()).To(HaveLen(10)) - Expect(workerctl.AllPaths()).To(ContainElement(workerctl.PathBackendInstall)) - Expect(workerctl.AllPaths()).To(ContainElement(workerctl.PathNodeStop)) + It("lists every verb this package names, so none can be dropped from the set", func() { + // The claim is bounded on purpose. Go constants are not enumerable, so + // nothing here can see a NEW constant that was never added to AllPaths; + // what this catches is an EXISTING verb going missing from it, which + // matters because the prefix check above and the worker's mounting spec + // both iterate AllPaths and would silently stop covering it. + Expect(workerctl.AllPaths()).To(ConsistOf( + workerctl.PathBackendInstall, + workerctl.PathBackendUpgrade, + workerctl.PathBackendList, + workerctl.PathBackendStop, + workerctl.PathBackendDelete, + workerctl.PathModelStop, + workerctl.PathModelUnload, + workerctl.PathModelDelete, + workerctl.PathModelsRunning, + workerctl.PathNodeStop, + )) }) It("gives each verb a distinct path", func() { diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index a057dbed1..09b485c65 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -186,7 +186,11 @@ Two of the routes stream. `POST /v1/control/backend/install` and `/v1/control/backend/upgrade` answer with `application/x-ndjson`: zero or more `{"progress":{...}}` lines carrying the same download-progress payload the per-op NATS subject carried, followed by exactly one `{"reply":{...}}` line, -which is always the last line on the body. An install that FAILS is still a +which is always the last line on the body. When the request carries an `op_id`, +the first progress line has phase `resolving` and is written before any gallery +work begins, so a cold install that spends minutes on a manifest is +distinguishable from a stream that is broken. Nothing publishes install progress +over NATS any more. An install that FAILS is still a `200` with a reply whose `success` is `false`. That is deliberate, and it is the same distinction the refusal table above draws: a non-2xx means the frontend could not get the request to the worker, which nothing may act on, while the