From aa5a1e47a49856b7d42f66f31da1df3c5fc1a6f9 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Mon, 20 Jul 2026 17:57:12 +0000 Subject: [PATCH] fix(modelartifacts): treat CIFS EACCES as lock contention, not failure flock(2) on CIFS/SMB returns EACCES when another client holds the lock: the kernel maps STATUS_LOCK_NOT_GRANTED and STATUS_FILE_LOCK_CONFLICT to -EACCES and never produces EWOULDBLOCK on that path. gofrs/flock only recognises EWOULDBLOCK as contention, so TryLockContext returned a bare "permission denied" and Ensure aborted. Both replicas then fell back to legacy loading, which makes the worker download the whole repo in-band inside LoadModel and blow the remote-load deadline. Replace TryLockContext with an explicit wait loop over a new Locker interface, classifying EWOULDBLOCK/EAGAIN/EACCES/EBUSY as contention. EACCES is ambiguous at the syscall boundary but not here: the lock file is already open O_CREATE|O_RDWR, so a real permission problem would have failed the open with an *fs.PathError, and flock(2) documents no EACCES on Linux at all. The wait is bounded (DefaultLockWait, overridable via WithLockWait), so even a misclassification degrades to a delay. On timeout the committed result is re-checked before reporting the new ErrLockContended, so a peer that finished the work still wins. Locker also exists so the contention path is testable without a network filesystem: nothing in CI can make flock(2) return EACCES on demand. Raise the fallback to error for a managedArtifactBackends backend, via a shared config.LogArtifactFallback used by both call sites. For those backends the legacy path is not graceful degradation, and the operator otherwise sees only a timeout with no causal link. The fallback stays non-fatal. Drop the os.Chmod(layout.Lock, 0o600) after acquisition: flock.New already creates the file 0600, and the chmod was gratuitous risk on a nounix mount that ignores modes. Fixes #10981 Signed-off-by: Ettore Di Giacinto Assisted-by: Claude Code:claude-opus-4-8[1m] [Read] [Edit] [Bash] --- core/config/model_artifact_fallback_test.go | 62 +++++++ core/config/model_artifact_inference.go | 21 +++ core/config/model_config_loader.go | 2 +- core/gallery/model_artifacts.go | 3 +- core/gallery/model_artifacts_fallback_test.go | 57 ++++++ pkg/modelartifacts/materializer.go | 147 +++++++++++++-- pkg/modelartifacts/materializer_lock_test.go | 169 ++++++++++++++++++ 7 files changed, 445 insertions(+), 16 deletions(-) create mode 100644 core/config/model_artifact_fallback_test.go create mode 100644 core/gallery/model_artifacts_fallback_test.go create mode 100644 pkg/modelartifacts/materializer_lock_test.go diff --git a/core/config/model_artifact_fallback_test.go b/core/config/model_artifact_fallback_test.go new file mode 100644 index 000000000000..d55c62370529 --- /dev/null +++ b/core/config/model_artifact_fallback_test.go @@ -0,0 +1,62 @@ +package config + +import ( + "bytes" + "context" + "log/slog" + "os" + "path/filepath" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/xlog" +) + +// For a backend in managedArtifactBackends the legacy path is not a graceful +// degradation: the backend consumes a snapshot directory, so falling back means +// it downloads the whole repo itself, in-band, inside LoadModel on every load. +// That reliably blows the remote-load deadline, and the operator sees only a +// timeout unless the cause was logged loudly enough to survive a default log +// level. #10981 traced a 57 GB in-band pull back to a warn logged 40 minutes +// earlier. +var _ = Describe("preload artifact fallback severity", func() { + const inferredConfig = ` +name: managed +backend: transformers +parameters: {model: owner/repo} +` + + var captured *bytes.Buffer + + BeforeEach(func() { + // Capture at error level so a warn emission is filtered out and the + // assertion fails on severity, not on wording. + captured = &bytes.Buffer{} + handler := slog.NewTextHandler(captured, &slog.HandlerOptions{Level: slog.LevelError}) + xlog.SetLogger(xlog.NewLoggerWithHandler(handler, xlog.LogLevelError)) + }) + + AfterEach(func() { + // xlog exposes no getter for the package logger, so restore the same + // default the suite entrypoint installs rather than the prior value. + xlog.SetLogger(xlog.NewLogger(xlog.LogLevel("info"), "text")) + }) + + It("logs at error, and names the in-band download, when a managed backend degrades", func() { + modelsPath := GinkgoT().TempDir() + Expect(os.WriteFile(filepath.Join(modelsPath, "managed.yaml"), []byte(inferredConfig), 0644)).To(Succeed()) + + fake := &companionMaterializer{fail: "owner/repo"} + loader := NewModelConfigLoader(modelsPath, WithArtifactMaterializer(fake)) + Expect(loader.LoadModelConfigsFromPath(modelsPath)).To(Succeed()) + + // Behavior is unchanged: the degradation is still non-fatal. + Expect(loader.PreloadWithContext(context.Background(), modelsPath)).To(Succeed()) + + logged := captured.String() + Expect(logged).To(ContainSubstring("managed")) + Expect(logged).To(ContainSubstring("transformers")) + Expect(logged).To(ContainSubstring("in-band")) + }) +}) diff --git a/core/config/model_artifact_inference.go b/core/config/model_artifact_inference.go index b9c90856e611..48a3238f8c80 100644 --- a/core/config/model_artifact_inference.go +++ b/core/config/model_artifact_inference.go @@ -7,6 +7,7 @@ import ( "strings" "github.com/mudler/LocalAI/pkg/modelartifacts" + "github.com/mudler/xlog" ) // managedArtifactBackends is the set of backends that load a model from a @@ -33,6 +34,26 @@ func IsManagedArtifactBackend(backend string) bool { return ok } +// LogArtifactFallback records a degradation to the legacy loading path. For a +// backend that consumes a snapshot *directory*, "legacy loading" is not a +// graceful degradation: the backend downloads the whole repo itself, in-band, +// inside LoadModel, on every load. That reliably exceeds the remote-load +// deadline and surfaces as a bare timeout, so it is logged at error to survive +// a default log level and give the timeout a named cause (#10981). +// +// Today every inferred artifact already comes from such a backend, but the gate +// is checked explicitly so widening PrimaryArtifactSpec later cannot silently +// promote an ordinary fallback to an error. +func LogArtifactFallback(model, backend string, err error) { + if IsManagedArtifactBackend(backend) { + xlog.Error("artifact materialization failed; the backend will download the model in-band on every load", + "model", model, "backend", backend, "error", err) + return + } + xlog.Warn("falling back to legacy model loading after artifact materialization failed", + "model", model, "backend", backend, "error", err) +} + // PrimaryArtifactSpec returns the managed primary artifact to materialize for // this config. The boolean return is false when the config should stay on the // legacy path. diff --git a/core/config/model_config_loader.go b/core/config/model_config_loader.go index eea1058d322d..bd9e93865035 100644 --- a/core/config/model_config_loader.go +++ b/core/config/model_config_loader.go @@ -490,7 +490,7 @@ func (bcl *ModelConfigLoader) preloadOne( result, err := bcl.artifactMaterializer.Ensure(ctx, modelPath, artifactSpec) if err != nil { if inferred { - xlog.Warn("falling back to legacy model loading after artifact materialization failed", "model", updated.Name, "error", err) + LogArtifactFallback(updated.Name, updated.Backend, err) managedPrimary = false } else { return ModelConfig{}, nil, err diff --git a/core/gallery/model_artifacts.go b/core/gallery/model_artifacts.go index fb302157c603..2fa46a0d31b4 100644 --- a/core/gallery/model_artifacts.go +++ b/core/gallery/model_artifacts.go @@ -10,7 +10,6 @@ import ( "github.com/mudler/LocalAI/core/config" "github.com/mudler/LocalAI/pkg/modelartifacts" - "github.com/mudler/xlog" ) type ArtifactMaterializer interface { @@ -54,7 +53,7 @@ func bindPrimaryArtifact(ctx context.Context, modelsPath string, typed *config.M result, err := materializer.Ensure(ctx, modelsPath, artifactSpec) if err != nil { if inferred { - xlog.Warn("falling back to legacy model loading after artifact materialization failed", "model", typed.Name, "error", err) + config.LogArtifactFallback(typed.Name, typed.Backend, err) return false, nil } return false, fmt.Errorf("materialize primary model artifact: %w", err) diff --git a/core/gallery/model_artifacts_fallback_test.go b/core/gallery/model_artifacts_fallback_test.go new file mode 100644 index 000000000000..9debaa938e49 --- /dev/null +++ b/core/gallery/model_artifacts_fallback_test.go @@ -0,0 +1,57 @@ +package gallery_test + +import ( + "bytes" + "context" + "errors" + "log/slog" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/core/gallery" + "github.com/mudler/LocalAI/pkg/system" + "github.com/mudler/xlog" +) + +// Install-time degradation has the same consequence as preload-time +// degradation: a managed backend that never gets its snapshot downloads the +// repo itself on every load. See #10981. +var _ = Describe("install artifact fallback severity", func() { + var captured *bytes.Buffer + + BeforeEach(func() { + // Capture at error level so a warn emission is filtered out and the + // assertion fails on severity, not on wording. + captured = &bytes.Buffer{} + handler := slog.NewTextHandler(captured, &slog.HandlerOptions{Level: slog.LevelError}) + xlog.SetLogger(xlog.NewLoggerWithHandler(handler, xlog.LogLevelError)) + }) + + AfterEach(func() { + // xlog exposes no getter for the package logger, so restore the same + // default the suite entrypoint installs rather than the prior value. + xlog.SetLogger(xlog.NewLogger(xlog.LogLevel("info"), "text")) + }) + + It("logs at error, and names the in-band download, when a managed backend degrades", func() { + state, err := system.GetSystemState(system.WithModelPath(GinkgoT().TempDir())) + Expect(err).NotTo(HaveOccurred()) + fake := &fakeArtifactMaterializer{err: errors.New("materialization refused")} + definition := &gallery.ModelConfig{Name: "legacy", ConfigFile: ` +backend: transformers +parameters: + model: owner/legacy +`} + + // Behavior is unchanged: the install still succeeds on the legacy path. + _, err = gallery.InstallModel(context.Background(), state, "legacy", definition, nil, nil, false, + gallery.WithArtifactMaterializer(fake)) + Expect(err).NotTo(HaveOccurred()) + + logged := captured.String() + Expect(logged).To(ContainSubstring("legacy")) + Expect(logged).To(ContainSubstring("transformers")) + Expect(logged).To(ContainSubstring("in-band")) + }) +}) diff --git a/pkg/modelartifacts/materializer.go b/pkg/modelartifacts/materializer.go index e17499c773e3..149b96d2ef41 100644 --- a/pkg/modelartifacts/materializer.go +++ b/pkg/modelartifacts/materializer.go @@ -6,12 +6,14 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "os" "path" "path/filepath" "strings" + "syscall" "time" "github.com/gofrs/flock" @@ -25,9 +27,39 @@ type SnapshotResolver interface { ResolveSnapshot(context.Context, hfapi.SnapshotRequest) (hfapi.Snapshot, error) } +// Locker is the cross-process exclusion primitive that keeps two replicas from +// materializing the same snapshot at once. It is an interface rather than a +// concrete *flock.Flock because the contention path has to be exercised without +// a network filesystem: no test environment can make flock(2) return the +// CIFS-specific errno this code must tolerate, and it is also the seam a +// database-backed lock would plug into for multi-node deployments. +type Locker interface { + TryLock() (bool, error) + Unlock() error +} + +// ErrLockContended reports that a peer held the artifact lock for the whole +// wait window and never published a usable snapshot. It is deliberately +// distinct from an acquisition failure: the work is in progress elsewhere, so +// callers can say so rather than blaming the model. +var ErrLockContended = errors.New("artifact lock is held by another process") + +const ( + // DefaultLockWait bounds how long Ensure waits for a peer replica. It is + // generous because the peer may legitimately be downloading tens of + // gigabytes, and waiting is strictly cheaper than the fallback, which makes + // the backend download the same repo in-band. + DefaultLockWait = 30 * time.Minute + + initialLockRetryInterval = 100 * time.Millisecond + maxLockRetryInterval = 5 * time.Second +) + type Manager struct { resolver SnapshotResolver huggingFaceToken string + newLocker func(string) Locker + lockWait time.Duration } type ManagerOption func(*Manager) @@ -43,8 +75,32 @@ func WithHuggingFaceToken(token string) ManagerOption { return func(manager *Manager) { manager.huggingFaceToken = token } } +// WithLocker overrides how the artifact lock is created. The default is an +// flock(2) lock on the shared models directory. +func WithLocker(factory func(path string) Locker) ManagerOption { + return func(manager *Manager) { + if factory != nil { + manager.newLocker = factory + } + } +} + +// WithLockWait bounds how long Ensure waits for a peer replica holding the +// artifact lock before giving up with ErrLockContended. +func WithLockWait(wait time.Duration) ManagerOption { + return func(manager *Manager) { + if wait > 0 { + manager.lockWait = wait + } + } +} + func NewManager(resolver SnapshotResolver, options ...ManagerOption) *Manager { - manager := &Manager{resolver: resolver} + manager := &Manager{ + resolver: resolver, + newLocker: func(path string) Locker { return flock.New(path) }, + lockWait: DefaultLockWait, + } for _, option := range options { option(manager) } @@ -147,25 +203,24 @@ func (m *Manager) Ensure(ctx context.Context, modelsPath string, spec Spec) (Res if err := os.MkdirAll(filepath.Dir(layout.Lock), 0o750); err != nil { return Result{}, err } - artifactLock := flock.New(layout.Lock) - locked, err := artifactLock.TryLockContext(ctx, 100*time.Millisecond) - if err != nil { - return Result{}, err - } - if !locked { - if err := ctx.Err(); err != nil { - return Result{}, err + artifactLock := m.newLocker(layout.Lock) + if err := m.acquireLock(ctx, artifactLock, layout.Lock); err != nil { + // A peer that held the lock for the whole window was very likely doing + // exactly this work. Its committed snapshot is the answer we wanted, so + // prefer it over reporting contention to a caller that would degrade to + // an in-band download. + if errors.Is(err, ErrLockContended) { + if cached, ok := committedResult(modelsPath, normalized); ok { + return cached, nil + } } - return Result{}, fmt.Errorf("artifact lock was not acquired") + return Result{}, err } defer func() { if err := artifactLock.Unlock(); err != nil { xlog.Warn("failed to unlock model artifact", "lock", layout.Lock, "error", err) } }() - if err := os.Chmod(layout.Lock, 0o600); err != nil { - return Result{}, err - } if cached, ok := committedResult(modelsPath, normalized); ok { return cached, nil } @@ -175,6 +230,72 @@ func (m *Manager) Ensure(ctx context.Context, modelsPath string, spec Spec) (Res return m.materializeLocked(ctx, modelsPath, normalized, snapshot, token, layout) } +// isLockContention reports whether a failed lock attempt means "somebody else +// holds it" rather than "this will never work". +// +// Only EWOULDBLOCK is portable, and it is the only errno gofrs/flock treats as +// contention. Network filesystems translate their own protocol status codes +// instead: CIFS/SMB maps STATUS_LOCK_NOT_GRANTED and STATUS_FILE_LOCK_CONFLICT +// to EACCES, and a busy share can surface EBUSY. Both look like hard errors and +// aborted materialization outright on a shared /models (#10981). +// +// EACCES is ambiguous at the syscall boundary, where it also spells "permission +// denied", but it is not ambiguous at this call site. The lock file was already opened +// O_CREATE|O_RDWR before we get here, so a genuine permission problem would +// have failed the open with an *fs.PathError naming the path. flock(2) itself +// documents no EACCES on Linux (EBADF, EINTR, EINVAL, ENOLCK, EWOULDBLOCK), so +// a bare EACCES from the lock call can only have come from a network +// filesystem's lock-conflict translation. The wait is bounded regardless, so +// even a misclassification degrades to a delay, not a hang. +func isLockContention(err error) bool { + return errors.Is(err, syscall.EWOULDBLOCK) || + errors.Is(err, syscall.EAGAIN) || + errors.Is(err, syscall.EACCES) || + errors.Is(err, syscall.EBUSY) +} + +// acquireLock blocks until the artifact lock is held, the context is done, or +// the wait window expires with ErrLockContended. +func (m *Manager) acquireLock(ctx context.Context, locker Locker, lockPath string) error { + wait := m.lockWait + if wait <= 0 { + wait = DefaultLockWait + } + deadline := time.Now().Add(wait) + interval := initialLockRetryInterval + waited := false + for { + if err := ctx.Err(); err != nil { + return err + } + locked, err := locker.TryLock() + if err != nil && !isLockContention(err) { + return err + } + if locked { + if waited { + xlog.Info("acquired the model artifact lock after waiting for another replica", "lock", lockPath) + } + return nil + } + if !waited { + waited = true + xlog.Info("another replica is materializing this model artifact; waiting for it", "lock", lockPath) + } + if time.Now().After(deadline) { + return fmt.Errorf("%w: %s", ErrLockContended, lockPath) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(interval): + } + if interval < maxLockRetryInterval { + interval = min(interval*2, maxLockRetryInterval) + } + } +} + func removeInvalidFinal(layout Layout) error { root, err := os.OpenRoot(layout.Root) if err != nil { diff --git a/pkg/modelartifacts/materializer_lock_test.go b/pkg/modelartifacts/materializer_lock_test.go new file mode 100644 index 000000000000..ef8706048617 --- /dev/null +++ b/pkg/modelartifacts/materializer_lock_test.go @@ -0,0 +1,169 @@ +package modelartifacts_test + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "net/http" + "net/http/httptest" + "sync" + "syscall" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + hfapi "github.com/mudler/LocalAI/pkg/huggingface-api" + "github.com/mudler/LocalAI/pkg/modelartifacts" +) + +// lockAttempt is one scripted outcome of Locker.TryLock. Attempts beyond the +// script acquire the lock. +type lockAttempt struct { + locked bool + err error +} + +type scriptedLocker struct { + mu sync.Mutex + script []lockAttempt + attempts int + unlocks int + onAttempt func(attempt int) +} + +func (s *scriptedLocker) TryLock() (bool, error) { + s.mu.Lock() + s.attempts++ + attempt := s.attempts + var outcome lockAttempt + if attempt <= len(s.script) { + outcome = s.script[attempt-1] + } else { + outcome = lockAttempt{locked: true} + } + callback := s.onAttempt + s.mu.Unlock() + if callback != nil { + callback(attempt) + } + return outcome.locked, outcome.err +} + +func (s *scriptedLocker) Unlock() error { + s.mu.Lock() + defer s.mu.Unlock() + s.unlocks++ + return nil +} + +func (s *scriptedLocker) attemptCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.attempts +} + +// singleFileResolver serves one small file so a full Ensure can run end to end +// without the test caring about download mechanics. +func singleFileResolver() *fakeSnapshotResolver { + weight := []byte("weight-bytes") + sum := sha256.Sum256(weight) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write(weight) + })) + DeferCleanup(server.Close) + return &fakeSnapshotResolver{snapshot: hfapi.Snapshot{ + Endpoint: "https://huggingface.co", Repo: "owner/repo", + RequestedRevision: "main", ResolvedRevision: "0123456789abcdef0123456789abcdef01234567", + Files: []hfapi.SnapshotFile{{ + Path: "model.safetensors", Size: int64(len(weight)), + LFSOID: hex.EncodeToString(sum[:]), URL: server.URL + "/model", + }}, + }} +} + +var specUnderTest = modelartifacts.Spec{Source: modelartifacts.Source{Type: "huggingface", Repo: "owner/repo"}} + +// CIFS/SMB maps STATUS_LOCK_NOT_GRANTED and STATUS_FILE_LOCK_CONFLICT to +// EACCES, so a contended flock(2) on a shared /models never produces the +// EWOULDBLOCK that gofrs/flock recognises. Treating that as a hard failure made +// both replicas abandon materialization and degrade to an in-band download of +// the whole repo inside LoadModel (#10981). +var _ = Describe("artifact lock contention on network filesystems", func() { + It("waits out CIFS-style EACCES contention and then materializes", func() { + locker := &scriptedLocker{script: []lockAttempt{ + {err: syscall.EACCES}, + {err: syscall.EACCES}, + }} + manager := modelartifacts.NewManager(singleFileResolver(), + modelartifacts.WithLocker(func(string) modelartifacts.Locker { return locker }), + modelartifacts.WithLockWait(10*time.Second)) + + result, err := manager.Ensure(context.Background(), GinkgoT().TempDir(), specUnderTest) + Expect(err).NotTo(HaveOccurred()) + Expect(result.RelativePath).To(HavePrefix(".artifacts/huggingface/")) + Expect(locker.attemptCount()).To(Equal(3)) + }) + + It("waits out an EWOULDBLOCK contention reported without an error", func() { + locker := &scriptedLocker{script: []lockAttempt{{}, {}}} + manager := modelartifacts.NewManager(singleFileResolver(), + modelartifacts.WithLocker(func(string) modelartifacts.Locker { return locker }), + modelartifacts.WithLockWait(10*time.Second)) + + _, err := manager.Ensure(context.Background(), GinkgoT().TempDir(), specUnderTest) + Expect(err).NotTo(HaveOccurred()) + Expect(locker.attemptCount()).To(Equal(3)) + }) + + It("does not retry a lock error that is not contention", func() { + locker := &scriptedLocker{script: []lockAttempt{{err: syscall.ENOLCK}}} + manager := modelartifacts.NewManager(singleFileResolver(), + modelartifacts.WithLocker(func(string) modelartifacts.Locker { return locker }), + modelartifacts.WithLockWait(10*time.Second)) + + _, err := manager.Ensure(context.Background(), GinkgoT().TempDir(), specUnderTest) + Expect(err).To(MatchError(syscall.ENOLCK)) + Expect(locker.attemptCount()).To(Equal(1)) + }) + + It("adopts the peer's snapshot when contention outlasts the wait window", func() { + modelsPath := GinkgoT().TempDir() + resolver := singleFileResolver() + // The "peer" replica commits the snapshot while we are still waiting, + // which is the whole point of waiting: the result exists even though we + // never held the lock ourselves. + peer := modelartifacts.NewManager(resolver, + modelartifacts.WithLocker(func(string) modelartifacts.Locker { return &scriptedLocker{} })) + locker := &scriptedLocker{ + script: []lockAttempt{{err: syscall.EACCES}, {err: syscall.EACCES}, {err: syscall.EACCES}, {err: syscall.EACCES}}, + onAttempt: func(attempt int) { + if attempt != 2 { + return + } + _, err := peer.Ensure(context.Background(), modelsPath, specUnderTest) + Expect(err).NotTo(HaveOccurred()) + }, + } + manager := modelartifacts.NewManager(resolver, + modelartifacts.WithLocker(func(string) modelartifacts.Locker { return locker }), + modelartifacts.WithLockWait(300*time.Millisecond)) + + result, err := manager.Ensure(context.Background(), modelsPath, specUnderTest) + Expect(err).NotTo(HaveOccurred()) + Expect(result.CacheHit).To(BeTrue()) + }) + + It("reports contention distinctly when the peer never commits", func() { + locker := &scriptedLocker{script: []lockAttempt{ + {err: syscall.EACCES}, {err: syscall.EACCES}, {err: syscall.EACCES}, + {err: syscall.EACCES}, {err: syscall.EACCES}, {err: syscall.EACCES}, + }} + manager := modelartifacts.NewManager(singleFileResolver(), + modelartifacts.WithLocker(func(string) modelartifacts.Locker { return locker }), + modelartifacts.WithLockWait(200*time.Millisecond)) + + _, err := manager.Ensure(context.Background(), GinkgoT().TempDir(), specUnderTest) + Expect(err).To(MatchError(modelartifacts.ErrLockContended)) + }) +})