Skip to content

Commit 3948b58

Browse files
committed
fix(distributed): worker stopBackend/isRunning resolve bare modelID to replica keys
PR #9583 changed the supervisor's process map key from `modelID` to `modelID#replicaIndex`, but the NATS lifecycle handlers kept passing the bare modelID: * `backend.stop` (subscribeLifecycleEvents): `s.stopBackend(req.Backend)` → `s.processes["Qwen3.6-..."]` missed (actual key is "...#0") → silent no-op. Admin "Unload model" clicks released VRAM via model.unload but left the gRPC process alive on its old port. Subsequent chats hit installBackend, found the leftover process, reused its address — and the UI reported "no models loaded" while the model kept responding. * `backend.delete` (subscribeLifecycleEvents): same map miss in `isRunning(req.Backend)` and `s.stopBackend(req.Backend)` — admin "Delete backend" deleted the binary while the process was still serving traffic. Add `resolveProcessKeys(id)`: exact match if `id` is a full processKey (stopAllBackends iterates the map and passes its own keys); prefix-match if `id` is bare (NATS handlers); empty if `id` contains `#` but doesn't match (no spurious fallback when the caller was explicit). stopBackend and isRunning now call it; stopBackend gets a new stopBackendExact helper for per-key cleanup. TDD: regression test fails without the fix (resolveProcessKeys doesn't exist; map lookup by bare name returns nothing). Tests pass post-fix. Reproduced live: registry row count was 0 for the model the user "Unloaded", chat still served by the leftover worker process. SmartRouter behavior is correct in itself — it falls through to scheduleAndLoad when no row exists; the bug was that the leftover process corrupted the install path. Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Assisted-by: claude-code:opus-4-7 [Edit] [Bash]
1 parent 5efbe84 commit 3948b58

2 files changed

Lines changed: 118 additions & 14 deletions

File tree

core/cli/worker.go

Lines changed: 67 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -502,33 +502,74 @@ func (s *backendSupervisor) startBackend(backend, backendPath string) (string, e
502502
return clientAddr, nil
503503
}
504504

505-
// stopBackend stops a specific backend's gRPC process.
506-
func (s *backendSupervisor) stopBackend(backend string) {
505+
// resolveProcessKeys turns a caller-supplied identifier into the set of
506+
// process map keys it refers to. PR #9583 changed s.processes to be keyed by
507+
// `modelID#replicaIndex`, but external NATS handlers still pass the bare
508+
// model ID — without this resolver, those lookups silently no-op'd, so
509+
// admin "Unload model" / "Delete backend" left the worker process alive.
510+
//
511+
// - Exact match wins. Callers that already know the full processKey
512+
// (stopAllBackends iterating its own map) get exactly that entry.
513+
// - Else, an identifier without `#` is treated as a model prefix and
514+
// every `id#N` replica is returned.
515+
// - An identifier that contains `#` but doesn't match anything returns
516+
// nothing — no spurious prefix fallback when the caller was explicit.
517+
func (s *backendSupervisor) resolveProcessKeys(id string) []string {
518+
s.mu.Lock()
519+
defer s.mu.Unlock()
520+
if _, ok := s.processes[id]; ok {
521+
return []string{id}
522+
}
523+
if strings.Contains(id, "#") {
524+
return nil
525+
}
526+
prefix := id + "#"
527+
var keys []string
528+
for k := range s.processes {
529+
if strings.HasPrefix(k, prefix) {
530+
keys = append(keys, k)
531+
}
532+
}
533+
return keys
534+
}
535+
536+
// stopBackend stops the backend process(es) matching the given identifier.
537+
// Accepts a bare modelID (stops every replica) or a full processKey
538+
// (stops just that replica).
539+
func (s *backendSupervisor) stopBackend(id string) {
540+
for _, key := range s.resolveProcessKeys(id) {
541+
s.stopBackendExact(key)
542+
}
543+
}
544+
545+
// stopBackendExact stops the process under exactly this key. Locking and
546+
// network I/O are split: the map mutation runs under the lock, the gRPC
547+
// Free() and proc.Stop() calls run after release so they don't block
548+
// other supervisor operations.
549+
func (s *backendSupervisor) stopBackendExact(key string) {
507550
s.mu.Lock()
508-
bp, ok := s.processes[backend]
551+
bp, ok := s.processes[key]
509552
if !ok || bp.proc == nil {
510553
s.mu.Unlock()
511554
return
512555
}
513-
// Clean up map and recycle port while holding lock
514-
delete(s.processes, backend)
556+
delete(s.processes, key)
515557
if _, portStr, err := net.SplitHostPort(bp.addr); err == nil {
516558
if p, err := strconv.Atoi(portStr); err == nil {
517559
s.freePorts = append(s.freePorts, p)
518560
}
519561
}
520562
s.mu.Unlock()
521563

522-
// Network I/O outside the lock
523564
client := grpc.NewClientWithToken(bp.addr, false, nil, false, s.cmd.RegistrationToken)
524-
xlog.Debug("Calling Free() before stopping backend", "backend", backend)
565+
xlog.Debug("Calling Free() before stopping backend", "backend", key)
525566
if err := client.Free(context.Background()); err != nil {
526-
xlog.Warn("Free() failed (best-effort)", "backend", backend, "error", err)
567+
xlog.Warn("Free() failed (best-effort)", "backend", key, "error", err)
527568
}
528569

529-
xlog.Info("Stopping backend process", "backend", backend, "addr", bp.addr)
570+
xlog.Info("Stopping backend process", "backend", key, "addr", bp.addr)
530571
if err := bp.proc.Stop(); err != nil {
531-
xlog.Error("Error stopping backend process", "backend", backend, "error", err)
572+
xlog.Error("Error stopping backend process", "backend", key, "error", err)
532573
}
533574
}
534575

@@ -557,12 +598,24 @@ func readLastLinesFromFile(path string, n int) string {
557598
return strings.Join(lines, "\n")
558599
}
559600

560-
// isRunning returns whether a specific backend process is currently running.
561-
func (s *backendSupervisor) isRunning(backend string) bool {
601+
// isRunning returns whether at least one backend process matching the given
602+
// identifier is currently running. Accepts a bare modelID (matches any
603+
// replica) or a full processKey (exact match). Callers like the
604+
// backend.delete pre-check rely on the bare-name path.
605+
func (s *backendSupervisor) isRunning(id string) bool {
606+
keys := s.resolveProcessKeys(id)
607+
if len(keys) == 0 {
608+
// Same lock-free zero-process check the caller would have done.
609+
return false
610+
}
562611
s.mu.Lock()
563612
defer s.mu.Unlock()
564-
bp, ok := s.processes[backend]
565-
return ok && bp.proc != nil && bp.proc.IsAlive()
613+
for _, key := range keys {
614+
if bp, ok := s.processes[key]; ok && bp.proc != nil && bp.proc.IsAlive() {
615+
return true
616+
}
617+
}
618+
return false
566619
}
567620

568621
// getAddr returns the gRPC address for a running backend, or empty string.

core/cli/worker_replica_test.go

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,4 +67,55 @@ var _ = Describe("Worker per-replica process keying", func() {
6767
Expect(labels).To(HaveKeyWithValue("node.replica-slots", "2"))
6868
})
6969
})
70+
71+
Describe("Process map lookup by bare model name", func() {
72+
// Regression: PR #9583 changed the supervisor's map key from
73+
// `modelID` to `modelID#replicaIndex`. The NATS backend.stop
74+
// handler kept passing the bare modelID, so the lookup silently
75+
// no-op'd — the worker process stayed alive after an admin
76+
// "Unload model" click, and subsequent chats kept being served
77+
// by the leftover process. The registry rows were gone, so the
78+
// UI reported "no models loaded" while the model kept
79+
// responding. resolveProcessKeys must turn a bare modelID into
80+
// the actual replica process keys so stop/isRunning find the
81+
// running processes.
82+
It("resolves a bare modelID to its replica process keys", func() {
83+
s := &backendSupervisor{
84+
processes: map[string]*backendProcess{
85+
"qwen3.6-35B#0": {addr: "127.0.0.1:50051"},
86+
"qwen3.6-35B#1": {addr: "127.0.0.1:50052"},
87+
"other-model#0": {addr: "127.0.0.1:50053"},
88+
},
89+
}
90+
keys := s.resolveProcessKeys("qwen3.6-35B")
91+
Expect(keys).To(ConsistOf("qwen3.6-35B#0", "qwen3.6-35B#1"),
92+
"bare modelID must match all replica process keys")
93+
94+
// Bare modelID for a model with no live processes returns nothing.
95+
Expect(s.resolveProcessKeys("not-loaded")).To(BeEmpty())
96+
97+
// Full processKey resolves to itself (per-replica callers stay precise).
98+
Expect(s.resolveProcessKeys("qwen3.6-35B#0")).To(ConsistOf("qwen3.6-35B#0"))
99+
100+
// A processKey that doesn't exist returns nothing — no spurious
101+
// prefix fallback when the caller was explicit.
102+
Expect(s.resolveProcessKeys("qwen3.6-35B#9")).To(BeEmpty())
103+
})
104+
105+
It("isRunning returns false when no replica matches", func() {
106+
// We can only test the not-found path without a real *process.Process
107+
// (IsAlive() requires PID introspection). That's enough to pin the
108+
// regression — pre-fix, isRunning("qwen3.6-35B") would always
109+
// return false because the map was keyed by "qwen3.6-35B#0".
110+
// Post-fix, isRunning calls resolveProcessKeys first, so the
111+
// per-replica lookup is exercised before the IsAlive probe.
112+
s := &backendSupervisor{processes: map[string]*backendProcess{}}
113+
Expect(s.isRunning("qwen3.6-35B")).To(BeFalse())
114+
// resolveProcessKeys finds the replica entries (the lookup contract
115+
// is what the backend.delete handler relies on); the IsAlive probe
116+
// itself is exercised by the integration path in distributed mode.
117+
s.processes["qwen3.6-35B#0"] = &backendProcess{addr: "127.0.0.1:50051"}
118+
Expect(s.resolveProcessKeys("qwen3.6-35B")).To(ConsistOf("qwen3.6-35B#0"))
119+
})
120+
})
70121
})

0 commit comments

Comments
 (0)