mirror of
https://github.com/mudler/LocalAI.git
synced 2026-11-05 20:13:23 -05:00
feat(studio): check the shared blob volume at startup
Without S3, nodes that use PostgreSQL metadata and the filesystem blob store must mount the same blob folder. Nothing checked this, so a volume that was not shared made files written on one frontend look missing on another while the index said they existed. The first node to start writes a random id in the blob root and records it in the database. Every node compares the two at startup and fails with a clear error on a mismatch or a missing id file. Assisted-by: Claude Code:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
This commit is contained in:
6 files changed
+212
-3
No files matched your search
@@ -228,6 +228,12 @@ func (a *Application) openWorkspaceIndex(root string, blobs blob.Store) (workspa
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
// Without S3, nodes that share the database must share the blob
|
||||
// folder too; a volume that is not shared loses files silently.
|
||||
if err := ws.VerifyBlobVolume(context.Background()); err != nil {
|
||||
_ = ws.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
return ws, ws, nil
|
||||
}
|
||||
ws, err := jsonstore.New(filepath.Join(root, "workspace"), blobs, jsonstore.Options{Admin: files.AdminPrincipal, MaxFileSize: studioMaxFileSize})
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/mudler/LocalAI/core/services/studio/workspace"
|
||||
)
|
||||
@@ -284,3 +285,37 @@ func (r contextReader) Read(p []byte) (int, error) {
|
||||
|
||||
// IsFilesystem distinguishes authoritative direct FS from an object-store cache.
|
||||
func (*fsStore) IsFilesystem() bool { return true }
|
||||
|
||||
// volumeMarker is the file at the blob root whose content names the volume.
|
||||
const volumeMarker = ".volume-id"
|
||||
|
||||
// ReadVolumeID returns the identity stored at the blob root, or "" when the
|
||||
// root has none yet.
|
||||
func (s *fsStore) ReadVolumeID(_ context.Context) (string, error) {
|
||||
b, err := os.ReadFile(filepath.Join(s.root, volumeMarker)) // #nosec G304 -- fixed name under the canonical root.
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return "", nil
|
||||
}
|
||||
return strings.TrimSpace(string(b)), err
|
||||
}
|
||||
|
||||
// WriteVolumeID stores id at the blob root. It fails with os.ErrExist when
|
||||
// the root already has an identity, so two writers cannot both claim it.
|
||||
func (s *fsStore) WriteVolumeID(_ context.Context, id string) error {
|
||||
f, err := os.OpenFile(filepath.Join(s.root, volumeMarker), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o400) // #nosec G304 -- fixed name under the canonical root.
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = f.WriteString(id + "\n")
|
||||
if err == nil {
|
||||
err = f.Sync()
|
||||
}
|
||||
if cerr := f.Close(); err == nil {
|
||||
err = cerr
|
||||
}
|
||||
if err != nil {
|
||||
_ = os.Remove(f.Name())
|
||||
return err
|
||||
}
|
||||
return s.syncDir(s.root)
|
||||
}
|
||||
@@ -74,3 +74,15 @@ func (s *Staging) Close() {
|
||||
}
|
||||
|
||||
var _ Stager = (*fsStore)(nil)
|
||||
|
||||
// Volume is an optional capability of a Store whose bytes live on a volume
|
||||
// that every node must share. The workspace keeps the identity in its
|
||||
// database and compares it with the one on the volume at startup.
|
||||
type Volume interface {
|
||||
// ReadVolumeID returns the identity stored on the volume, or "" if none.
|
||||
ReadVolumeID(ctx context.Context) (string, error)
|
||||
// WriteVolumeID stores id. It fails with os.ErrExist if an identity is there.
|
||||
WriteVolumeID(ctx context.Context, id string) error
|
||||
}
|
||||
|
||||
var _ Volume = (*fsStore)(nil)
|
||||
@@ -14,9 +14,12 @@ package gormstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -221,3 +224,68 @@ func mapUnique(err error) error {
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// ErrVolumeMismatch means this node does not see the blob volume that the
|
||||
// database was set up with.
|
||||
var ErrVolumeMismatch = errors.New("workspace: the blob folder is not the one the database expects")
|
||||
|
||||
const volumeKey = "blob_volume"
|
||||
|
||||
// VerifyBlobVolume checks, on a shared filesystem, that this node sees the
|
||||
// same blob folder as the other nodes of the database. The first node to start
|
||||
// gives the folder a random identity and records it in the database; every
|
||||
// node compares the two. A volume that is not shared would make a file written
|
||||
// through one frontend look missing on another. It does nothing for other
|
||||
// setups.
|
||||
func (s *Workspace) VerifyBlobVolume(ctx context.Context) error {
|
||||
vol, ok := s.blobs.(blob.Volume)
|
||||
if !s.sharedFilesystem || !ok {
|
||||
return nil
|
||||
}
|
||||
fileID, err := vol.ReadVolumeID(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
dbID, err := s.volumeID(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if fileID == "" && dbID == "" {
|
||||
var raw [16]byte
|
||||
if _, err := rand.Read(raw[:]); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := vol.WriteVolumeID(ctx, hex.EncodeToString(raw[:])); err != nil && !errors.Is(err, os.ErrExist) {
|
||||
return err
|
||||
}
|
||||
if fileID, err = vol.ReadVolumeID(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if dbID == "" && fileID != "" {
|
||||
// Several nodes may race here; the first insert wins and the others
|
||||
// read it back below.
|
||||
if _, err := s.db.ExecContext(ctx, `INSERT INTO studio_ws_meta (key, value) VALUES (?, ?) ON CONFLICT (key) DO NOTHING`, volumeKey, fileID); err != nil {
|
||||
return err
|
||||
}
|
||||
if dbID, err = s.volumeID(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if fileID != dbID {
|
||||
if fileID == "" {
|
||||
return fmt.Errorf("%w: it has no identity file, so it is another or an emptied folder; every node must mount the same blob folder", ErrVolumeMismatch)
|
||||
}
|
||||
return fmt.Errorf("%w: its identity differs from the database; every node must mount the same blob folder", ErrVolumeMismatch)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Workspace) volumeID(ctx context.Context) (string, error) {
|
||||
var id string
|
||||
err := s.db.QueryRowContext(ctx, `SELECT value FROM studio_ws_meta WHERE key = ?`, volumeKey).Scan(&id)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return "", nil
|
||||
}
|
||||
return id, err
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
package gormstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/mudler/LocalAI/core/http/auth"
|
||||
"github.com/mudler/LocalAI/core/services/studio/workspace/blob"
|
||||
)
|
||||
|
||||
var _ = Describe("VerifyBlobVolume", func() {
|
||||
ctx := context.Background()
|
||||
var (
|
||||
db *gorm.DB
|
||||
root string
|
||||
)
|
||||
// node opens a store the way one frontend would. SQLite stands in for
|
||||
// PostgreSQL, so the shared-filesystem mode is switched on by hand.
|
||||
node := func(dir string) *Workspace {
|
||||
GinkgoHelper()
|
||||
fs, err := blob.NewFS(dir)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
s, err := New(db, fs, Options{Admin: "admin"})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
s.sharedFilesystem = true
|
||||
return s
|
||||
}
|
||||
|
||||
BeforeEach(func() {
|
||||
var err error
|
||||
db, err = auth.OpenSQLiteFile(filepath.Join(GinkgoT().TempDir(), "ws.db"))
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
DeferCleanup(func() {
|
||||
if raw, err := db.DB(); err == nil {
|
||||
_ = raw.Close()
|
||||
}
|
||||
})
|
||||
root = filepath.Join(GinkgoT().TempDir(), "blobs")
|
||||
})
|
||||
|
||||
It("names a new volume and accepts it again on every node that mounts it", func() {
|
||||
Expect(node(root).VerifyBlobVolume(ctx)).To(Succeed())
|
||||
Expect(node(root).VerifyBlobVolume(ctx)).To(Succeed())
|
||||
})
|
||||
|
||||
It("refuses a node whose blob folder is another volume", func() {
|
||||
Expect(node(root).VerifyBlobVolume(ctx)).To(Succeed())
|
||||
other := node(filepath.Join(GinkgoT().TempDir(), "other"))
|
||||
Expect(other.VerifyBlobVolume(ctx)).To(MatchError(ErrVolumeMismatch))
|
||||
})
|
||||
|
||||
It("refuses a node whose folder has another identity", func() {
|
||||
first := node(filepath.Join(GinkgoT().TempDir(), "a"))
|
||||
Expect(first.VerifyBlobVolume(ctx)).To(Succeed())
|
||||
// A second volume that was named before it met this database.
|
||||
second := node(filepath.Join(GinkgoT().TempDir(), "b"))
|
||||
vol := second.blobs.(blob.Volume)
|
||||
Expect(vol.WriteVolumeID(ctx, "someone-else")).To(Succeed())
|
||||
Expect(second.VerifyBlobVolume(ctx)).To(MatchError(ErrVolumeMismatch))
|
||||
})
|
||||
|
||||
It("adopts the identity of a volume when the database has none", func() {
|
||||
fs, err := blob.NewFS(root)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(fs.(blob.Volume).WriteVolumeID(ctx, "kept")).To(Succeed())
|
||||
Expect(node(root).VerifyBlobVolume(ctx)).To(Succeed())
|
||||
id, err := node(root).volumeID(ctx)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(id).To(Equal("kept"))
|
||||
})
|
||||
|
||||
It("does nothing without a shared filesystem", func() {
|
||||
s := node(root)
|
||||
s.sharedFilesystem = false
|
||||
Expect(s.VerifyBlobVolume(ctx)).To(Succeed())
|
||||
id, err := s.volumeID(ctx)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(id).To(BeEmpty())
|
||||
})
|
||||
})
|
||||
@@ -1694,7 +1694,8 @@ example sets it up with Garage. Without S3, Files uses the existing
|
||||
`DataPath/chats/blobs` directly. All Files participants must then mount the same
|
||||
backing data directory and use the same PostgreSQL metadata database. See
|
||||
[Shared filesystem for Files](#shared-filesystem-for-files) for the operator
|
||||
contract; startup does not verify shared volume identity.
|
||||
contract. With PostgreSQL, startup checks that every node sees the same blob
|
||||
folder and stops with an error when it does not.
|
||||
|
||||
## Storage
|
||||
|
||||
@@ -1947,8 +1948,11 @@ live in the selected backend. Per-instance caches and scratch are disposable.
|
||||
For filesystem storage, every Files consumer must mount the same existing data
|
||||
directory/backing storage and use the same PostgreSQL database: frontends, the
|
||||
add-on owner and optional chat workers. Inference-only nodes need no Files mount.
|
||||
Equal path strings do not prove shared volume identity, and startup does not
|
||||
verify shared storage or datastore identity. No local probe certifies a deployment.
|
||||
Equal path strings do not prove shared volume identity. With PostgreSQL, the
|
||||
first node to start writes a random `.volume-id` file in `DataPath/chats/blobs`
|
||||
and records the same id in the database; every node compares the two at startup
|
||||
and refuses to start on a mismatch or when the file is missing. This catches a
|
||||
volume that is not shared. It does not prove that the storage is coherent.
|
||||
Other data-path contents do not thereby become cluster-safe.
|
||||
|
||||
Keep the canonical backing directory service-controlled, with compatible UID/GID
|
||||
|
||||
Reference in new issue
Block a user