Files
LocalAI/core/services/galleryop/backends.go
LocalAI [bot] 8d6548c0b9 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>
2026-05-25 17:28:14 +02:00

213 lines
7.4 KiB
Go

package galleryop
import (
"context"
"errors"
"fmt"
"path/filepath"
"strings"
"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"
"github.com/mudler/LocalAI/pkg/utils"
"github.com/mudler/xlog"
)
func (g *GalleryService) backendHandler(op *ManagementOp[gallery.GalleryBackend, any], systemState *system.SystemState) error {
utils.ResetDownloadTimers()
// Dedup check in distributed mode — skip if another instance is already processing this element
if g.galleryStore != nil && op.GalleryElementName != "" {
dup, err := g.galleryStore.FindDuplicate(op.GalleryElementName)
if err == nil && dup != nil && dup.ID != op.ID {
g.UpdateStatus(op.ID, &OpStatus{
Processed: true,
Message: fmt.Sprintf("already being processed by another instance (op %s)", dup.ID),
})
return nil
}
}
// Check if already cancelled
if op.Context != nil {
select {
case <-op.Context.Done():
g.UpdateStatus(op.ID, &OpStatus{
Cancelled: true,
Processed: true,
Message: "cancelled",
GalleryElementName: op.GalleryElementName,
})
return op.Context.Err()
default:
}
}
g.UpdateStatus(op.ID, &OpStatus{Message: fmt.Sprintf("processing backend: %s", op.GalleryElementName), Progress: 0, Cancellable: true})
// displayDownload displays the download progress
progressCallback := func(fileName string, current string, total string, percentage float64) {
// Check for cancellation during progress updates
if op.Context != nil {
select {
case <-op.Context.Done():
return
default:
}
}
g.UpdateStatus(op.ID, &OpStatus{Message: fmt.Sprintf(processingMessage, fileName, total, current), FileName: fileName, Progress: percentage, TotalFileSize: total, DownloadedFileSize: current, Cancellable: true})
utils.DisplayDownloadFunction(fileName, current, total, percentage)
}
ctx := op.Context
if ctx == nil {
ctx = context.Background()
}
var err error
if op.Upgrade {
err = g.backendManager.UpgradeBackend(ctx, op.GalleryElementName, progressCallback)
} else if op.Delete {
err = g.backendManager.DeleteBackend(op.GalleryElementName)
} else {
err = g.backendManager.InstallBackend(ctx, op, progressCallback)
// Update GalleryElementName for status tracking if a name was derived
if op.ExternalName != "" {
op.GalleryElementName = op.ExternalName
}
}
if err != nil {
// Check if error is due to cancellation
if op.Context != nil && errors.Is(err, op.Context.Err()) {
g.UpdateStatus(op.ID, &OpStatus{
Cancelled: true,
Processed: true,
Message: "cancelled",
GalleryElementName: op.GalleryElementName,
})
return err
}
if errors.Is(err, ErrWorkerStillInstalling) {
// Soft failure: at least one worker timed out replying but is
// still running the install in the background. Mark the op as
// processed with a non-error message so the admin UI shows a
// yellow in-progress state rather than red. The reconciler's
// next pass will reconcile the actual outcome via backend.list.
xlog.Info("worker still installing in background", "backend", op.GalleryElementName, "error", err)
g.UpdateStatus(op.ID, &OpStatus{
Processed: true,
GalleryElementName: op.GalleryElementName,
Message: fmt.Sprintf("backend %s: worker still installing in background; reconciler will confirm completion (%v)", op.GalleryElementName, err),
Cancellable: false,
})
return nil
}
xlog.Error("error installing backend", "error", err, "backend", op.GalleryElementName)
if !op.Delete {
// If we didn't install the backend, we need to make sure we don't have a leftover directory
gallery.DeleteBackendFromSystem(systemState, op.GalleryElementName)
}
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,
Processed: true,
GalleryElementName: op.GalleryElementName,
Message: "completed",
Progress: 100,
Cancellable: false})
return nil
}
// InstallExternalBackend installs a backend from an external source (OCI image, URL, or path).
// This method contains the logic to detect the input type and call the appropriate installation function.
// It can be used by both CLI and Web UI for installing backends from external sources.
func InstallExternalBackend(ctx context.Context, galleries []config.Gallery, systemState *system.SystemState, modelLoader *model.ModelLoader, downloadStatus func(string, string, string, float64), backend, name, alias string, requireIntegrity bool) error {
uri := downloader.URI(backend)
switch {
case uri.LooksLikeDir():
if name == "" { // infer it from the path
name = filepath.Base(backend)
}
xlog.Info("Installing backend from path", "backend", backend, "name", name)
if err := gallery.InstallBackend(ctx, systemState, modelLoader, &gallery.GalleryBackend{
Metadata: gallery.Metadata{
Name: name,
},
Alias: alias,
URI: backend,
}, downloadStatus, requireIntegrity); err != nil {
return fmt.Errorf("error installing backend %s: %w", backend, err)
}
case uri.LooksLikeOCI() && !uri.LooksLikeOCIFile():
if name == "" {
return fmt.Errorf("specifying a name is required for OCI images")
}
xlog.Info("Installing backend from OCI image", "backend", backend, "name", name)
if err := gallery.InstallBackend(ctx, systemState, modelLoader, &gallery.GalleryBackend{
Metadata: gallery.Metadata{
Name: name,
},
Alias: alias,
URI: backend,
}, downloadStatus, requireIntegrity); err != nil {
return fmt.Errorf("error installing backend %s: %w", backend, err)
}
case uri.LooksLikeOCIFile():
derivedName, err := uri.FilenameFromUrl()
if err != nil {
return fmt.Errorf("failed to get filename from URL: %w", err)
}
// strip extension if any
derivedName = strings.TrimSuffix(derivedName, filepath.Ext(derivedName))
// Use provided name if available, otherwise use derived name
if name == "" {
name = derivedName
}
xlog.Info("Installing backend from OCI image", "backend", backend, "name", name)
if err := gallery.InstallBackend(ctx, systemState, modelLoader, &gallery.GalleryBackend{
Metadata: gallery.Metadata{
Name: name,
},
Alias: alias,
URI: backend,
}, downloadStatus, requireIntegrity); err != nil {
return fmt.Errorf("error installing backend %s: %w", backend, err)
}
default:
// Treat as gallery backend name
if name != "" || alias != "" {
return fmt.Errorf("specifying a name or alias is not supported for gallery backends")
}
err := gallery.InstallBackendFromGallery(ctx, galleries, systemState, modelLoader, backend, downloadStatus, true, requireIntegrity)
if err != nil {
return fmt.Errorf("error installing backend %s: %w", backend, err)
}
}
return nil
}