feat(distributed/worker): add per-backend mutex helper to backendSupervisor

Different backend names lock independently; same backend serializes. This
is the synchronization primitive used by the upcoming concurrent install
handler — without it, wrapping the NATS callback in a goroutine would
race the gallery directory when two requests target the same backend.

Assisted-by: Claude:claude-opus-4-7
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

core/cli/worker.go

@@ -417,6 +417,14 @@ type backendSupervisor struct {
 	processes map[string]*backendProcess // key: backend name
 	nextPort  int                        // next available port for new backends
 	freePorts []int                      // ports freed by stopBackend, reused before nextPort
+
+	// backendLocks serializes gallery operations against the same on-disk
+	// artifact. Two installs of different backends on the same worker run
+	// concurrently (their handlers are each in a goroutine); two operations
+	// on the same backend (install vs upgrade, or two parallel installs of
+	// the same not-yet-cached backend) are serialized here so the gallery
+	// download path doesn't race itself on the same directory.
+	backendLocks map[string]*sync.Mutex
 }
 
 // startBackend starts a gRPC backend process on a dynamically allocated port.
@@ -792,6 +800,24 @@ func (s *backendSupervisor) findBackend(backend string) string {
 	return ""
 }
 
+// lockBackend returns a release function for a per-backend mutex. Different
+// backend names lock independently. The first caller for a name allocates
+// the mutex under s.mu; subsequent callers for the same name reuse it.
+func (s *backendSupervisor) lockBackend(name string) func() {
+	s.mu.Lock()
+	if s.backendLocks == nil {
+		s.backendLocks = make(map[string]*sync.Mutex)
+	}
+	m, ok := s.backendLocks[name]
+	if !ok {
+		m = &sync.Mutex{}
+		s.backendLocks[name] = m
+	}
+	s.mu.Unlock()
+	m.Lock()
+	return m.Unlock
+}
+
 // subscribeLifecycleEvents subscribes to NATS backend lifecycle events.
 func (s *backendSupervisor) subscribeLifecycleEvents() {
 	// backend.install — install backend + start gRPC process (request-reply)

core/cli/worker_concurrency_test.go

@@ -0,0 +1,72 @@
+package cli
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("backendSupervisor.lockBackend", func() {
+	It("serializes operations on the same backend name", func() {
+		s := &backendSupervisor{processes: map[string]*backendProcess{}}
+
+		var inflight, peak int32
+		var wg sync.WaitGroup
+		for i := 0; i < 4; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				release := s.lockBackend("llama-cpp")
+				defer release()
+
+				now := atomic.AddInt32(&inflight, 1)
+				for {
+					p := atomic.LoadInt32(&peak)
+					if now <= p || atomic.CompareAndSwapInt32(&peak, p, now) {
+						break
+					}
+				}
+				time.Sleep(20 * time.Millisecond)
+				atomic.AddInt32(&inflight, -1)
+			}()
+		}
+		wg.Wait()
+
+		Expect(atomic.LoadInt32(&peak)).To(Equal(int32(1)),
+			"only one goroutine should hold the per-backend lock at a time")
+	})
+
+	It("allows different backend names to run in parallel", func() {
+		s := &backendSupervisor{processes: map[string]*backendProcess{}}
+
+		var inflight, peak int32
+		var wg sync.WaitGroup
+		names := []string{"llama-cpp", "vllm", "whisper", "speaker-recognition"}
+		for _, n := range names {
+			n := n
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				release := s.lockBackend(n)
+				defer release()
+
+				now := atomic.AddInt32(&inflight, 1)
+				for {
+					p := atomic.LoadInt32(&peak)
+					if now <= p || atomic.CompareAndSwapInt32(&peak, p, now) {
+						break
+					}
+				}
+				time.Sleep(50 * time.Millisecond)
+				atomic.AddInt32(&inflight, -1)
+			}()
+		}
+		wg.Wait()
+
+		Expect(atomic.LoadInt32(&peak)).To(BeNumerically(">=", int32(2)),
+			"distinct backends should be able to run concurrently")
+	})
+})