fix(distributed): sync gallery OpCache + caches across frontend replicas (#9983)

When the LocalAI frontend deployment is scaled past one replica, the UI's
/api/operations poll round-robins between pods. Each pod kept the OpCache
(galleryID->jobID), OpStatus map, and the post-install in-memory caches
(ModelConfigLoader, UpgradeChecker) purely in-process. Reads never
consulted PostgreSQL or NATS even though writes already published to PG.
Symptoms:

- A user installing a model on replica A saw the operation card flicker
  in and out as the load balancer alternated.
- The Models page re-fetched the whole gallery on every flicker because
  useEffect([operations.length]) re-fires when the count changes.
- A chat completion that landed on replica B after the install completed
  on replica A failed to find the new model — B's ModelConfigLoader was
  still the old one because nothing told it to reload from disk.
- The UpgradeChecker 6-hour cache stayed stale on peer replicas after a
  backend upgrade, so /api/backends/upgrades kept surfacing an upgrade
  that had already shipped.

Mirror the jobs Dispatcher pattern for gallery ops:

- OpCache learns SetMessagingClient/SetGalleryStore + a Start(ctx) that
  hydrates from PostgreSQL and subscribes to gallery.opcache.{start,end}.
  Set/SetBackend now upsert cache_key + is_backend_op on the gallery_
  operations row and broadcast OpCacheEvent so peers merge it in. The
  hydrate path uses a new GalleryStore.ListActive() (status in {pending,
  downloading, processing} and updated within 30 min).
- GalleryService.SubscribeBroadcasts wires a SubjectGalleryProgress-
  Wildcard subscriber that calls a new lock-light mergeStatus into the
  local statuses map, plus a SubjectGalleryCancelWildcard subscriber that
  runs the locally-registered cancel func. Hydrate() restores active rows
  from PostgreSQL on startup so a freshly-started replica is not
  observably empty mid-install. CancelOperation tolerates the cancel func
  living on a different replica and publishes anyway.
- modelHandler and backendHandler publish on the new
  SubjectCacheInvalidateModels / SubjectCacheInvalidateBackends after
  a successful install/delete/upgrade. SubscribeBroadcasts wires peers
  to refresh: OnModelsChanged (re-runs LoadModelConfigsFromPath) and
  OnBackendOpCompleted (re-triggers UpgradeChecker). The originating
  replica reloads inline so it never enters the broadcast handler.
- OpStatus.Error (an error interface) flat-marshalled to "{}" over JSON,
  so a failed install replicated to a peer arrived with a nil error and
  the UI's failure banner never appeared. Add MarshalJSON/UnmarshalJSON
  via an opStatusWire shim that round-trips Error as a string.
- UpdateStatus and CancelOperation now drop the mutex before publishing
  to NATS or persisting to PostgreSQL. The wildcard subscriber's
  mergeStatus loops back into the same service on the publishing replica
  and would deadlock otherwise; this also prevents future PG round-trips
  from stalling concurrent readers on every progress tick.

Tests cover the OpStatus error round-trip, OpCache propagation through a
shared in-memory bus, OpCache PostgreSQL hydration (active-only),
GalleryService progress + cancel broadcast, Nodes preservation across a
peer's bare progress tick, GalleryService hydration from PG, and the
two cache-invalidation broadcasts (models + backends). 44 specs total
in galleryop; routes/operations specs and jobs/agents suites still pass.


Assisted-by: claude-code:claude-opus-4-7

Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
Co-authored-by: Ettore Di Giacinto <mudler@localai.io>

Author: localai-bot

core/application/startup.go

@@ -15,6 +15,7 @@ import (
 	"github.com/mudler/LocalAI/core/http/auth"
 	"github.com/mudler/LocalAI/core/services/galleryop"
 	"github.com/mudler/LocalAI/core/services/jobs"
+	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/LocalAI/core/services/monitoring"
 	"github.com/mudler/LocalAI/core/services/nodes"
 	"github.com/mudler/LocalAI/core/services/routing/admission"
@@ -312,6 +313,30 @@ func New(opts ...config.AppOption) (*Application, error) {
 				}
 				application.galleryService.SetGalleryStore(distSvc.DistStores.Gallery)
 			}
+			// Hydrate from the store first so the wildcard subscriber finds an
+			// already-populated statuses map for any operations still in flight
+			// on a peer replica.
+			if err := application.galleryService.Hydrate(); err != nil {
+				xlog.Warn("Gallery service hydrate failed", "error", err)
+			}
+			// Bind cache-invalidation handler before SubscribeBroadcasts so the
+			// first inbound event is already routed. Peer replicas install a
+			// model and broadcast on SubjectCacheInvalidateModels; this
+			// callback re-runs LoadModelConfigsFromPath so a subsequent chat
+			// completion that load-balances onto this replica finds the new
+			// config. The originating replica reloads inline in modelHandler
+			// and never enters this path.
+			gs := application.galleryService
+			sys := options.SystemState
+			cfgLoaderOpts := options.ToConfigLoaderOptions()
+			gs.OnModelsChanged = func(_ messaging.CacheInvalidateEvent) {
+				if err := application.ModelConfigLoader().LoadModelConfigsFromPath(sys.Model.ModelsPath, cfgLoaderOpts...); err != nil {
+					xlog.Warn("Failed to reload model configs after peer invalidation", "error", err)
+				}
+			}
+			if err := application.galleryService.SubscribeBroadcasts(); err != nil {
+				xlog.Warn("Gallery service subscribe failed", "error", err)
+			}
 			// Wire distributed model/backend managers so delete propagates to workers
 			application.galleryService.SetModelManager(
 				nodes.NewDistributedModelManager(options, application.modelLoader, distSvc.Unloader),

core/http/app.go

@@ -367,6 +367,20 @@ func API(application *application.Application) (*echo.Echo, error) {
 	var opcache *galleryop.OpCache
 	if !application.ApplicationConfig().DisableWebUI {
 		opcache = galleryop.NewOpCache(application.GalleryService())
+		// In distributed mode, wire the NATS client + gallery store so this
+		// replica's OpCache stays in sync with peers — without this the
+		// /api/operations endpoint returns whatever this single replica
+		// happened to admit, and a load-balanced UI poll alternates between
+		// "operation visible" and "operation gone" between replicas.
+		if d := application.Distributed(); d != nil {
+			opcache.SetMessagingClient(d.Nats)
+			if d.DistStores != nil && d.DistStores.Gallery != nil {
+				opcache.SetGalleryStore(d.DistStores.Gallery)
+			}
+			if err := opcache.Start(application.ApplicationConfig().Context); err != nil {
+				xlog.Warn("OpCache distributed subscribe failed; running standalone", "error", err)
+			}
+		}
 	}
 
 	mcpMw := auth.RequireFeature(application.AuthDB(), auth.FeatureMCP)

core/services/distributed/gallery.go

@@ -6,16 +6,25 @@ import (
 
 	"github.com/google/uuid"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 // GalleryOperationRecord tracks model/backend download operations in PostgreSQL.
+//
+// CacheKey and IsBackendOp mirror the in-memory OpCache held by each frontend
+// replica. They are written when a request first lands so a freshly-started
+// (or freshly-routed-to) replica can rebuild its OpCache from this table
+// instead of returning an empty `/api/operations` payload while the real
+// operation is still in flight on a peer.
 type GalleryOperationRecord struct {
 	ID                 string    `gorm:"primaryKey;size:36" json:"id"`
 	UserID             string    `gorm:"index;size:36" json:"user_id,omitempty"`
 	GalleryElementName string    `gorm:"size:255" json:"gallery_element_name"`
-	OpType             string    `gorm:"size:32" json:"op_type"`                // "model_install", "model_delete", "backend_install"
-	Status             string    `gorm:"size:32;default:pending" json:"status"` // pending, downloading, processing, completed, failed, cancelled
-	Progress           float64   `json:"progress"`                              // 0.0 to 1.0
+	CacheKey           string    `gorm:"index;size:512" json:"cache_key,omitempty"` // OpCache key (galleryID or node:<id>:<backend>)
+	IsBackendOp        bool      `json:"is_backend_op"`                             // true if installed via SetBackend
+	OpType             string    `gorm:"size:32" json:"op_type"`                    // "model_install", "model_delete", "backend_install"
+	Status             string    `gorm:"size:32;default:pending" json:"status"`     // pending, downloading, processing, completed, failed, cancelled
+	Progress           float64   `json:"progress"`                                  // 0.0 to 1.0
 	Message            string    `gorm:"type:text" json:"message,omitempty"`
 	Error              string    `gorm:"type:text" json:"error,omitempty"`
 	FileName           string    `gorm:"size:512" json:"file_name,omitempty"`
@@ -27,6 +36,12 @@ type GalleryOperationRecord struct {
 	UpdatedAt          time.Time `json:"updated_at"`
 }
 
+// activeStatuses lists the gallery_operations.status values that represent an
+// operation a replica should still surface via /api/operations. Hydration and
+// the dedup lookup share this set so the two paths never disagree about what
+// "still active" means.
+var activeStatuses = []string{"pending", "downloading", "processing"}
+
 func (GalleryOperationRecord) TableName() string { return "gallery_operations" }
 
 // GalleryStore manages gallery operation state in PostgreSQL.
@@ -42,14 +57,26 @@ func NewGalleryStore(db *gorm.DB) (*GalleryStore, error) {
 	return &GalleryStore{db: db}, nil
 }
 
-// Create stores a new gallery operation.
+// Create stores a new gallery operation. Tolerates a row already existing
+// for this ID — OpCache.Set may have written a placeholder row via
+// UpsertCacheKey before the galleryop service goroutine called Create, and
+// in that case we want to fill in the descriptive columns (gallery element
+// name, op type, status) rather than fail with a primary-key conflict.
+// CacheKey and IsBackendOp are intentionally not in DoUpdates so the
+// placeholder's values win.
 func (s *GalleryStore) Create(op *GalleryOperationRecord) error {
 	if op.ID == "" {
 		op.ID = uuid.New().String()
 	}
 	op.CreatedAt = time.Now()
 	op.UpdatedAt = op.CreatedAt
-	return s.db.Create(op).Error
+	return s.db.Clauses(clause.OnConflict{
+		Columns: []clause.Column{{Name: "id"}},
+		DoUpdates: clause.AssignmentColumns([]string{
+			"gallery_element_name", "op_type", "status",
+			"frontend_id", "user_id", "cancellable", "updated_at",
+		}),
+	}).Create(op).Error
 }
 
 // UpdateProgress updates progress for an operation.
@@ -93,14 +120,55 @@ func (s *GalleryStore) List(status string) ([]GalleryOperationRecord, error) {
 	return ops, q.Find(&ops).Error
 }
 
+// ListActive returns operations still considered in-flight — used by replicas
+// to rehydrate their in-memory OpCache + statuses on startup. Stale records
+// (older than 30 minutes without an update) are excluded so a crashed peer's
+// orphaned rows never resurrect on a healthy replica; the existing CleanStale
+// reaper eventually marks them failed.
+func (s *GalleryStore) ListActive() ([]GalleryOperationRecord, error) {
+	var ops []GalleryOperationRecord
+	staleCutoff := time.Now().Add(-30 * time.Minute)
+	err := s.db.Where("status IN ? AND updated_at > ?", activeStatuses, staleCutoff).
+		Order("created_at DESC").Find(&ops).Error
+	return ops, err
+}
+
+// UpsertCacheKey records the in-memory OpCache key + IsBackendOp flag on the
+// gallery_operations row, creating the row if it does not exist yet.
+//
+// Why upsert: OpCache.Set is called by the HTTP admission handler before the
+// galleryop service goroutine processes the operation and calls Create. If
+// OpCache wrote with a plain Updates() those columns would silently be lost
+// in the window between the two, so peer replicas hydrating in that window
+// would still rebuild an empty OpCache. Upsert closes that window.
+func (s *GalleryStore) UpsertCacheKey(id, cacheKey string, isBackend bool) error {
+	now := time.Now()
+	rec := GalleryOperationRecord{
+		ID:          id,
+		CacheKey:    cacheKey,
+		IsBackendOp: isBackend,
+		Status:      "pending",
+		CreatedAt:   now,
+		UpdatedAt:   now,
+	}
+	return s.db.Clauses(clause.OnConflict{
+		Columns: []clause.Column{{Name: "id"}},
+		DoUpdates: clause.Assignments(map[string]any{
+			"cache_key":     cacheKey,
+			"is_backend_op": isBackend,
+			"updated_at":    now,
+		}),
+	}).Create(&rec).Error
+}
+
 // FindDuplicate checks if another instance is already downloading the same element.
 // Only considers records updated within the last 30 minutes as active — older
 // in-progress records are assumed to be stale (crashed instance).
 func (s *GalleryStore) FindDuplicate(elementName string) (*GalleryOperationRecord, error) {
 	var op GalleryOperationRecord
 	staleCutoff := time.Now().Add(-30 * time.Minute)
 	err := s.db.Where("gallery_element_name = ? AND status IN ? AND updated_at > ?", elementName,
-		[]string{"pending", "downloading", "processing"}, staleCutoff).First(&op).Error
+		activeStatuses, staleCutoff).First(&op).Error
 	if err != nil {
 		return nil, err
 	}
@@ -118,8 +186,7 @@ func (s *GalleryStore) Cancel(id string) error {
 func (s *GalleryStore) CleanStale(age time.Duration) error {
 	cutoff := time.Now().Add(-age)
 	return s.db.Model(&GalleryOperationRecord{}).
-		Where("updated_at < ? AND status IN ?", cutoff,
-			[]string{"pending", "downloading", "processing"}).
+		Where("updated_at < ? AND status IN ?", cutoff, activeStatuses).
 		Updates(map[string]any{
 			"status":     "failed",
 			"error":      "stale operation cleaned up on startup",

core/services/galleryop/backends.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/mudler/LocalAI/core/config"
 	"github.com/mudler/LocalAI/core/gallery"
+	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/LocalAI/pkg/downloader"
 	"github.com/mudler/LocalAI/pkg/model"
 	"github.com/mudler/LocalAI/pkg/system"
@@ -114,6 +115,21 @@ func (g *GalleryService) backendHandler(op *ManagementOp[gallery.GalleryBackend,
 		return err
 	}
 
+	// Tell peer replicas that the backend set has changed. UpgradeChecker
+	// caches upgrade-available bits for 6 hours, so without this peers would
+	// keep advertising an upgrade for a backend that already moved.
+	opName := "install"
+	switch {
+	case op.Delete:
+		opName = "delete"
+	case op.Upgrade:
+		opName = "upgrade"
+	}
+	g.publishCacheInvalidate(messaging.SubjectCacheInvalidateBackends, messaging.CacheInvalidateEvent{
+		Element: op.GalleryElementName,
+		Op:      opName,
+	})
+
 	g.UpdateStatus(op.ID,
 		&OpStatus{
 			Deletion:           op.Delete,