fix(distributed): sync gallery OpCache + statuses across frontend replicas

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) and OpStatus map purely in-process, and reads never
consulted PostgreSQL or NATS even though writes already published to both.
A user installing a model on replica A saw the operation card flicker in
and out as the load balancer alternated, and the Models page re-fetched
the whole gallery on every flicker because its useEffect watches the
operations count.

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.
- 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, and GalleryService hydration from PG. The
existing routes/operations specs and jobs/agents suites still pass.

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

Author: mudler

core/application/startup.go

@@ -312,6 +312,15 @@ 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)
+			}
+			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",