Skip to content

Commit aa5a1e4

Browse files
committed
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 <mudler@localai.io> Assisted-by: Claude Code:claude-opus-4-8[1m] [Read] [Edit] [Bash]
1 parent f381844 commit aa5a1e4

7 files changed

Lines changed: 445 additions & 16 deletions

File tree

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
package config
2+
3+
import (
4+
"bytes"
5+
"context"
6+
"log/slog"
7+
"os"
8+
"path/filepath"
9+
10+
. "github.com/onsi/ginkgo/v2"
11+
. "github.com/onsi/gomega"
12+
13+
"github.com/mudler/xlog"
14+
)
15+
16+
// For a backend in managedArtifactBackends the legacy path is not a graceful
17+
// degradation: the backend consumes a snapshot directory, so falling back means
18+
// it downloads the whole repo itself, in-band, inside LoadModel on every load.
19+
// That reliably blows the remote-load deadline, and the operator sees only a
20+
// timeout unless the cause was logged loudly enough to survive a default log
21+
// level. #10981 traced a 57 GB in-band pull back to a warn logged 40 minutes
22+
// earlier.
23+
var _ = Describe("preload artifact fallback severity", func() {
24+
const inferredConfig = `
25+
name: managed
26+
backend: transformers
27+
parameters: {model: owner/repo}
28+
`
29+
30+
var captured *bytes.Buffer
31+
32+
BeforeEach(func() {
33+
// Capture at error level so a warn emission is filtered out and the
34+
// assertion fails on severity, not on wording.
35+
captured = &bytes.Buffer{}
36+
handler := slog.NewTextHandler(captured, &slog.HandlerOptions{Level: slog.LevelError})
37+
xlog.SetLogger(xlog.NewLoggerWithHandler(handler, xlog.LogLevelError))
38+
})
39+
40+
AfterEach(func() {
41+
// xlog exposes no getter for the package logger, so restore the same
42+
// default the suite entrypoint installs rather than the prior value.
43+
xlog.SetLogger(xlog.NewLogger(xlog.LogLevel("info"), "text"))
44+
})
45+
46+
It("logs at error, and names the in-band download, when a managed backend degrades", func() {
47+
modelsPath := GinkgoT().TempDir()
48+
Expect(os.WriteFile(filepath.Join(modelsPath, "managed.yaml"), []byte(inferredConfig), 0644)).To(Succeed())
49+
50+
fake := &companionMaterializer{fail: "owner/repo"}
51+
loader := NewModelConfigLoader(modelsPath, WithArtifactMaterializer(fake))
52+
Expect(loader.LoadModelConfigsFromPath(modelsPath)).To(Succeed())
53+
54+
// Behavior is unchanged: the degradation is still non-fatal.
55+
Expect(loader.PreloadWithContext(context.Background(), modelsPath)).To(Succeed())
56+
57+
logged := captured.String()
58+
Expect(logged).To(ContainSubstring("managed"))
59+
Expect(logged).To(ContainSubstring("transformers"))
60+
Expect(logged).To(ContainSubstring("in-band"))
61+
})
62+
})

core/config/model_artifact_inference.go

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"strings"
88

99
"github.com/mudler/LocalAI/pkg/modelartifacts"
10+
"github.com/mudler/xlog"
1011
)
1112

1213
// managedArtifactBackends is the set of backends that load a model from a
@@ -33,6 +34,26 @@ func IsManagedArtifactBackend(backend string) bool {
3334
return ok
3435
}
3536

37+
// LogArtifactFallback records a degradation to the legacy loading path. For a
38+
// backend that consumes a snapshot *directory*, "legacy loading" is not a
39+
// graceful degradation: the backend downloads the whole repo itself, in-band,
40+
// inside LoadModel, on every load. That reliably exceeds the remote-load
41+
// deadline and surfaces as a bare timeout, so it is logged at error to survive
42+
// a default log level and give the timeout a named cause (#10981).
43+
//
44+
// Today every inferred artifact already comes from such a backend, but the gate
45+
// is checked explicitly so widening PrimaryArtifactSpec later cannot silently
46+
// promote an ordinary fallback to an error.
47+
func LogArtifactFallback(model, backend string, err error) {
48+
if IsManagedArtifactBackend(backend) {
49+
xlog.Error("artifact materialization failed; the backend will download the model in-band on every load",
50+
"model", model, "backend", backend, "error", err)
51+
return
52+
}
53+
xlog.Warn("falling back to legacy model loading after artifact materialization failed",
54+
"model", model, "backend", backend, "error", err)
55+
}
56+
3657
// PrimaryArtifactSpec returns the managed primary artifact to materialize for
3758
// this config. The boolean return is false when the config should stay on the
3859
// legacy path.

core/config/model_config_loader.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -490,7 +490,7 @@ func (bcl *ModelConfigLoader) preloadOne(
490490
result, err := bcl.artifactMaterializer.Ensure(ctx, modelPath, artifactSpec)
491491
if err != nil {
492492
if inferred {
493-
xlog.Warn("falling back to legacy model loading after artifact materialization failed", "model", updated.Name, "error", err)
493+
LogArtifactFallback(updated.Name, updated.Backend, err)
494494
managedPrimary = false
495495
} else {
496496
return ModelConfig{}, nil, err

core/gallery/model_artifacts.go

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,6 @@ import (
1010

1111
"github.com/mudler/LocalAI/core/config"
1212
"github.com/mudler/LocalAI/pkg/modelartifacts"
13-
"github.com/mudler/xlog"
1413
)
1514

1615
type ArtifactMaterializer interface {
@@ -54,7 +53,7 @@ func bindPrimaryArtifact(ctx context.Context, modelsPath string, typed *config.M
5453
result, err := materializer.Ensure(ctx, modelsPath, artifactSpec)
5554
if err != nil {
5655
if inferred {
57-
xlog.Warn("falling back to legacy model loading after artifact materialization failed", "model", typed.Name, "error", err)
56+
config.LogArtifactFallback(typed.Name, typed.Backend, err)
5857
return false, nil
5958
}
6059
return false, fmt.Errorf("materialize primary model artifact: %w", err)
Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
package gallery_test
2+
3+
import (
4+
"bytes"
5+
"context"
6+
"errors"
7+
"log/slog"
8+
9+
. "github.com/onsi/ginkgo/v2"
10+
. "github.com/onsi/gomega"
11+
12+
"github.com/mudler/LocalAI/core/gallery"
13+
"github.com/mudler/LocalAI/pkg/system"
14+
"github.com/mudler/xlog"
15+
)
16+
17+
// Install-time degradation has the same consequence as preload-time
18+
// degradation: a managed backend that never gets its snapshot downloads the
19+
// repo itself on every load. See #10981.
20+
var _ = Describe("install artifact fallback severity", func() {
21+
var captured *bytes.Buffer
22+
23+
BeforeEach(func() {
24+
// Capture at error level so a warn emission is filtered out and the
25+
// assertion fails on severity, not on wording.
26+
captured = &bytes.Buffer{}
27+
handler := slog.NewTextHandler(captured, &slog.HandlerOptions{Level: slog.LevelError})
28+
xlog.SetLogger(xlog.NewLoggerWithHandler(handler, xlog.LogLevelError))
29+
})
30+
31+
AfterEach(func() {
32+
// xlog exposes no getter for the package logger, so restore the same
33+
// default the suite entrypoint installs rather than the prior value.
34+
xlog.SetLogger(xlog.NewLogger(xlog.LogLevel("info"), "text"))
35+
})
36+
37+
It("logs at error, and names the in-band download, when a managed backend degrades", func() {
38+
state, err := system.GetSystemState(system.WithModelPath(GinkgoT().TempDir()))
39+
Expect(err).NotTo(HaveOccurred())
40+
fake := &fakeArtifactMaterializer{err: errors.New("materialization refused")}
41+
definition := &gallery.ModelConfig{Name: "legacy", ConfigFile: `
42+
backend: transformers
43+
parameters:
44+
model: owner/legacy
45+
`}
46+
47+
// Behavior is unchanged: the install still succeeds on the legacy path.
48+
_, err = gallery.InstallModel(context.Background(), state, "legacy", definition, nil, nil, false,
49+
gallery.WithArtifactMaterializer(fake))
50+
Expect(err).NotTo(HaveOccurred())
51+
52+
logged := captured.String()
53+
Expect(logged).To(ContainSubstring("legacy"))
54+
Expect(logged).To(ContainSubstring("transformers"))
55+
Expect(logged).To(ContainSubstring("in-band"))
56+
})
57+
})

pkg/modelartifacts/materializer.go

Lines changed: 134 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -6,12 +6,14 @@ import (
66
"crypto/sha256"
77
"encoding/hex"
88
"encoding/json"
9+
"errors"
910
"fmt"
1011
"io"
1112
"os"
1213
"path"
1314
"path/filepath"
1415
"strings"
16+
"syscall"
1517
"time"
1618

1719
"github.com/gofrs/flock"
@@ -25,9 +27,39 @@ type SnapshotResolver interface {
2527
ResolveSnapshot(context.Context, hfapi.SnapshotRequest) (hfapi.Snapshot, error)
2628
}
2729

30+
// Locker is the cross-process exclusion primitive that keeps two replicas from
31+
// materializing the same snapshot at once. It is an interface rather than a
32+
// concrete *flock.Flock because the contention path has to be exercised without
33+
// a network filesystem: no test environment can make flock(2) return the
34+
// CIFS-specific errno this code must tolerate, and it is also the seam a
35+
// database-backed lock would plug into for multi-node deployments.
36+
type Locker interface {
37+
TryLock() (bool, error)
38+
Unlock() error
39+
}
40+
41+
// ErrLockContended reports that a peer held the artifact lock for the whole
42+
// wait window and never published a usable snapshot. It is deliberately
43+
// distinct from an acquisition failure: the work is in progress elsewhere, so
44+
// callers can say so rather than blaming the model.
45+
var ErrLockContended = errors.New("artifact lock is held by another process")
46+
47+
const (
48+
// DefaultLockWait bounds how long Ensure waits for a peer replica. It is
49+
// generous because the peer may legitimately be downloading tens of
50+
// gigabytes, and waiting is strictly cheaper than the fallback, which makes
51+
// the backend download the same repo in-band.
52+
DefaultLockWait = 30 * time.Minute
53+
54+
initialLockRetryInterval = 100 * time.Millisecond
55+
maxLockRetryInterval = 5 * time.Second
56+
)
57+
2858
type Manager struct {
2959
resolver SnapshotResolver
3060
huggingFaceToken string
61+
newLocker func(string) Locker
62+
lockWait time.Duration
3163
}
3264

3365
type ManagerOption func(*Manager)
@@ -43,8 +75,32 @@ func WithHuggingFaceToken(token string) ManagerOption {
4375
return func(manager *Manager) { manager.huggingFaceToken = token }
4476
}
4577

78+
// WithLocker overrides how the artifact lock is created. The default is an
79+
// flock(2) lock on the shared models directory.
80+
func WithLocker(factory func(path string) Locker) ManagerOption {
81+
return func(manager *Manager) {
82+
if factory != nil {
83+
manager.newLocker = factory
84+
}
85+
}
86+
}
87+
88+
// WithLockWait bounds how long Ensure waits for a peer replica holding the
89+
// artifact lock before giving up with ErrLockContended.
90+
func WithLockWait(wait time.Duration) ManagerOption {
91+
return func(manager *Manager) {
92+
if wait > 0 {
93+
manager.lockWait = wait
94+
}
95+
}
96+
}
97+
4698
func NewManager(resolver SnapshotResolver, options ...ManagerOption) *Manager {
47-
manager := &Manager{resolver: resolver}
99+
manager := &Manager{
100+
resolver: resolver,
101+
newLocker: func(path string) Locker { return flock.New(path) },
102+
lockWait: DefaultLockWait,
103+
}
48104
for _, option := range options {
49105
option(manager)
50106
}
@@ -147,25 +203,24 @@ func (m *Manager) Ensure(ctx context.Context, modelsPath string, spec Spec) (Res
147203
if err := os.MkdirAll(filepath.Dir(layout.Lock), 0o750); err != nil {
148204
return Result{}, err
149205
}
150-
artifactLock := flock.New(layout.Lock)
151-
locked, err := artifactLock.TryLockContext(ctx, 100*time.Millisecond)
152-
if err != nil {
153-
return Result{}, err
154-
}
155-
if !locked {
156-
if err := ctx.Err(); err != nil {
157-
return Result{}, err
206+
artifactLock := m.newLocker(layout.Lock)
207+
if err := m.acquireLock(ctx, artifactLock, layout.Lock); err != nil {
208+
// A peer that held the lock for the whole window was very likely doing
209+
// exactly this work. Its committed snapshot is the answer we wanted, so
210+
// prefer it over reporting contention to a caller that would degrade to
211+
// an in-band download.
212+
if errors.Is(err, ErrLockContended) {
213+
if cached, ok := committedResult(modelsPath, normalized); ok {
214+
return cached, nil
215+
}
158216
}
159-
return Result{}, fmt.Errorf("artifact lock was not acquired")
217+
return Result{}, err
160218
}
161219
defer func() {
162220
if err := artifactLock.Unlock(); err != nil {
163221
xlog.Warn("failed to unlock model artifact", "lock", layout.Lock, "error", err)
164222
}
165223
}()
166-
if err := os.Chmod(layout.Lock, 0o600); err != nil {
167-
return Result{}, err
168-
}
169224
if cached, ok := committedResult(modelsPath, normalized); ok {
170225
return cached, nil
171226
}
@@ -175,6 +230,72 @@ func (m *Manager) Ensure(ctx context.Context, modelsPath string, spec Spec) (Res
175230
return m.materializeLocked(ctx, modelsPath, normalized, snapshot, token, layout)
176231
}
177232

233+
// isLockContention reports whether a failed lock attempt means "somebody else
234+
// holds it" rather than "this will never work".
235+
//
236+
// Only EWOULDBLOCK is portable, and it is the only errno gofrs/flock treats as
237+
// contention. Network filesystems translate their own protocol status codes
238+
// instead: CIFS/SMB maps STATUS_LOCK_NOT_GRANTED and STATUS_FILE_LOCK_CONFLICT
239+
// to EACCES, and a busy share can surface EBUSY. Both look like hard errors and
240+
// aborted materialization outright on a shared /models (#10981).
241+
//
242+
// EACCES is ambiguous at the syscall boundary, where it also spells "permission
243+
// denied", but it is not ambiguous at this call site. The lock file was already opened
244+
// O_CREATE|O_RDWR before we get here, so a genuine permission problem would
245+
// have failed the open with an *fs.PathError naming the path. flock(2) itself
246+
// documents no EACCES on Linux (EBADF, EINTR, EINVAL, ENOLCK, EWOULDBLOCK), so
247+
// a bare EACCES from the lock call can only have come from a network
248+
// filesystem's lock-conflict translation. The wait is bounded regardless, so
249+
// even a misclassification degrades to a delay, not a hang.
250+
func isLockContention(err error) bool {
251+
return errors.Is(err, syscall.EWOULDBLOCK) ||
252+
errors.Is(err, syscall.EAGAIN) ||
253+
errors.Is(err, syscall.EACCES) ||
254+
errors.Is(err, syscall.EBUSY)
255+
}
256+
257+
// acquireLock blocks until the artifact lock is held, the context is done, or
258+
// the wait window expires with ErrLockContended.
259+
func (m *Manager) acquireLock(ctx context.Context, locker Locker, lockPath string) error {
260+
wait := m.lockWait
261+
if wait <= 0 {
262+
wait = DefaultLockWait
263+
}
264+
deadline := time.Now().Add(wait)
265+
interval := initialLockRetryInterval
266+
waited := false
267+
for {
268+
if err := ctx.Err(); err != nil {
269+
return err
270+
}
271+
locked, err := locker.TryLock()
272+
if err != nil && !isLockContention(err) {
273+
return err
274+
}
275+
if locked {
276+
if waited {
277+
xlog.Info("acquired the model artifact lock after waiting for another replica", "lock", lockPath)
278+
}
279+
return nil
280+
}
281+
if !waited {
282+
waited = true
283+
xlog.Info("another replica is materializing this model artifact; waiting for it", "lock", lockPath)
284+
}
285+
if time.Now().After(deadline) {
286+
return fmt.Errorf("%w: %s", ErrLockContended, lockPath)
287+
}
288+
select {
289+
case <-ctx.Done():
290+
return ctx.Err()
291+
case <-time.After(interval):
292+
}
293+
if interval < maxLockRetryInterval {
294+
interval = min(interval*2, maxLockRetryInterval)
295+
}
296+
}
297+
}
298+
178299
func removeInvalidFinal(layout Layout) error {
179300
root, err := os.OpenRoot(layout.Root)
180301
if err != nil {

0 commit comments

Comments
 (0)