diff --git a/core/application/startup.go b/core/application/startup.go index 0de052083..1252914d3 100644 --- a/core/application/startup.go +++ b/core/application/startup.go @@ -1,6 +1,7 @@ package application import ( + "context" "crypto/rand" "encoding/hex" "fmt" @@ -41,6 +42,7 @@ import ( func New(opts ...config.AppOption) (*Application, error) { options := config.NewApplicationConfig(opts...) + cleanupWorkspace := workspaceStartupCleanup(options) // Store a copy of the startup config (env/CLI only, before file // loading): the settings endpoint uses it to tell env-provided API @@ -68,6 +70,12 @@ func New(opts ...config.AppOption) (*Application, error) { trace.ConfigureBackendTracePersistence(options.DataPath) application := newApplication(options) application.startupConfig = &startupConfigCopy + startupComplete := false + defer func() { + if !startupComplete { + cleanupWorkspace(application) + } + }() xlog.Info("Starting LocalAI", "threads", options.Threads, "modelsPath", options.SystemState.Model.ModelsPath) xlog.Info("LocalAI version", "version", internal.PrintableVersion()) @@ -415,7 +423,9 @@ func New(opts ...config.AppOption) (*Application, error) { } // The chats service needs the cluster message bus, so it starts after distributed wiring. - application.initStudioFiles() + if err := application.initStudioFiles(); err != nil { + return nil, err + } application.initStudioService() if err := coreStartup.InstallModels(options.Context, application.GalleryService(), options.Galleries, options.BackendGalleries, options.SystemState, application.ModelLoader(), options.EnforcePredownloadScans, options.AutoloadBackendGalleries, options.RequireBackendIntegrity, nil, options.ModelsURL...); err != nil { @@ -608,6 +618,7 @@ func New(opts ...config.AppOption) (*Application, error) { application.markStartupComplete() xlog.Info("core/startup process completed!") + startupComplete = true return application, nil } @@ -819,3 +830,24 @@ func ensureAuthDBDirs(options *config.ApplicationConfig) error { } return nil } + +// workspaceStartupCleanup owns startup resources when workspace initialization +// can fail fatally: configured S3 (with or without auth), or PostgreSQL metadata. +// Never cancel the caller's context or change legacy JSON/SQLite without S3. +func workspaceStartupCleanup(options *config.ApplicationConfig) func(*Application) { + postgres := options.Auth.Enabled && (strings.HasPrefix(options.Auth.DatabaseURL, "postgres://") || strings.HasPrefix(options.Auth.DatabaseURL, "postgresql://")) + if options.Distributed.StorageURL == "" && !postgres { + return func(*Application) {} + } + ctx, cancel := context.WithCancel(options.Context) + options.Context = ctx + return func(a *Application) { + cancel() // stop background work before shutting down its dependencies + _ = a.Shutdown() + if a.authDB != nil { + if db, err := a.authDB.DB(); err == nil { + _ = db.Close() + } + } + } +} diff --git a/core/application/studio_files.go b/core/application/studio_files.go index 4528bf4bd..2bf520dda 100644 --- a/core/application/studio_files.go +++ b/core/application/studio_files.go @@ -2,6 +2,7 @@ package application import ( "context" + "fmt" "io" "os" "path/filepath" @@ -9,9 +10,9 @@ import ( "github.com/mudler/xlog" - "github.com/mudler/LocalAI/core/config" "github.com/mudler/LocalAI/core/services/addons" "github.com/mudler/LocalAI/core/services/addons/packages" + "github.com/mudler/LocalAI/core/services/storage" "github.com/mudler/LocalAI/core/services/studio/builtinskills" "github.com/mudler/LocalAI/core/services/studio/files" "github.com/mudler/LocalAI/core/services/studio/textindex" @@ -50,24 +51,38 @@ func (a *Application) StudioFiles() *files.Service { // SQLite file or PostgreSQL) the tree lives there, in the GORM store. With no // database it is a folder under the data path with a JSON index, in the JSON // store; a build without cgo and without any build tag has it too. The bytes -// live in the object store of a cluster, and otherwise in a folder under the -// data path. With no data path there is nowhere to keep files, so they are -// off. Moving files from one index to the other is not supported. -func (a *Application) initStudioFiles() { - dp := a.applicationConfig.DataPath - dist := a.Distributed() - if reason := clusterFilesBlocker(a.applicationConfig, dist != nil && dist.Store != nil); reason != "" { - a.studioFiles = files.Disabled(reason) - return +// live in configured S3 for every role, otherwise at DataPath/chats/blobs. +// Moving files between backends is not supported automatically. +func (a *Application) initStudioFiles() (result error) { + return a.initStudioFilesWithS3(func(ctx context.Context, cfg storage.S3Config) (storage.ObjectStore, error) { + return storage.NewS3Store(ctx, cfg) + }) +} + +func (a *Application) initStudioFilesWithS3(newS3 func(context.Context, storage.S3Config) (storage.ObjectStore, error)) (result error) { + objects, err := a.workspaceObjects(newS3) + if err != nil { + return err } - if dp == "" && (a.authDB == nil || dist == nil || dist.Store == nil) { + strict := objects != nil || (a.authDB != nil && a.authDB.Dialector.Name() == "postgres") + defer func() { + if result != nil && a.studioFilesTemp != "" { + _ = os.RemoveAll(a.studioFilesTemp) + a.studioFilesTemp = "" + } + }() + dp := a.applicationConfig.DataPath + if dp == "" && (a.authDB == nil || objects == nil) { a.studioFiles = files.Disabled("There is no data path, so files are off. Start LocalAI with a data path to keep files.") - return + return nil } var tmpCache string if dp == "" { // a cluster without a data path: the blob cache is a temporary folder dir, err := os.MkdirTemp("", "localai-files-*") if err != nil { + if strict { + return fmt.Errorf("workspace scratch: %w", err) + } a.studioFiles = files.Disabled("Files are off: there is no folder for their cache.") return } @@ -79,19 +94,24 @@ func (a *Application) initStudioFiles() { } var blobs blob.Store - var err error - if dist != nil && dist.Store != nil { - blobs = blob.NewObjectStore(dist.Store, filepath.Join(root, "blobcache")) + if objects != nil { + blobs = blob.NewObjectStore(objects, filepath.Join(root, "blobcache")) } else { blobs, err = blob.NewFS(filepath.Join(root, "blobs")) } if err != nil { + if strict { + return fmt.Errorf("workspace blob store: %w", err) + } xlog.Warn("Files disabled: cannot open the blob store", "error", err) a.studioFiles = files.Disabled("Files are off: their storage could not be opened.") return } ws, closer, err := a.openWorkspaceIndex(root, blobs) if err != nil { + if strict { + return fmt.Errorf("workspace index: %w", err) + } xlog.Warn("Files disabled: cannot start the workspace", "error", err) a.studioFiles = files.Disabled("Files are off: the workspace could not be started.") return @@ -103,6 +123,9 @@ func (a *Application) initStudioFiles() { if closer != nil { _ = closer.Close() } + if strict { + return fmt.Errorf("workspace folders: %w", err) + } return } // The office skills ship with the server. A copy that fails leaves the @@ -113,6 +136,7 @@ func (a *Application) initStudioFiles() { a.studioFiles, a.studioFilesIndex = f, closer a.initStudioTextIndex(f, root) a.syncAddonSkills(f) + return nil } // initStudioTextIndex builds the text index of the documents in the files and @@ -193,18 +217,6 @@ func addonSkillSync(f *files.Service) addons.SkillSync { } } -// clusterFilesBlocker returns why files cannot work in this cluster, or "" when -// they can. The bytes of a file go to the cluster's object store so that every -// frontend reads what another wrote. Without S3 the object store is a folder -// on the frontend itself, and each frontend would keep its own copy of every -// file, so files are off instead of silently wrong. -func clusterFilesBlocker(cfg *config.ApplicationConfig, clustered bool) string { - if clustered && cfg.Distributed.StorageURL == "" { - return "Files are off: in a cluster they need S3 object storage (the storage URL setting), because without it each frontend would keep its own copy of every file." - } - return "" -} - // openWorkspaceIndex opens the index of the workspace: the GORM store on the // auth database when there is one, and the JSON store in a folder under root // otherwise. The returned closer is nil for the shared database, which the diff --git a/core/application/studio_files_test.go b/core/application/studio_files_test.go index e95b4cc5b..94b5ffdec 100644 --- a/core/application/studio_files_test.go +++ b/core/application/studio_files_test.go @@ -57,22 +57,3 @@ var _ = Describe("studio files at startup", func() { Expect(err).To(HaveOccurred()) // nothing was created in the working directory }) }) - -// The bytes of a file go to the cluster's object store. Without S3 that store -// is a folder of one frontend, so a cluster without S3 has no files. -var _ = Describe("files in a cluster", func() { - It("are off without S3, and say why", func() { - cfg := &config.ApplicationConfig{} - Expect(clusterFilesBlocker(cfg, true)).To(And(ContainSubstring("S3"), ContainSubstring("own copy"))) - }) - - It("work with S3", func() { - cfg := &config.ApplicationConfig{} - cfg.Distributed.StorageURL = "http://s3.example:9000" - Expect(clusterFilesBlocker(cfg, true)).To(BeEmpty()) - }) - - It("are not limited outside a cluster", func() { - Expect(clusterFilesBlocker(&config.ApplicationConfig{}, false)).To(BeEmpty()) - }) -}) diff --git a/core/application/studio_shared_files.go b/core/application/studio_shared_files.go new file mode 100644 index 000000000..ab50ef746 --- /dev/null +++ b/core/application/studio_shared_files.go @@ -0,0 +1,35 @@ +// SPDX-License-Identifier: MIT +package application + +import ( + "context" + "fmt" + + "github.com/mudler/LocalAI/core/services/storage" +) + +// workspaceObjects resolves workspace bytes only; model/input staging is unchanged. +// The constructor is local to this call so tests need neither network nor global hooks. +func (a *Application) workspaceObjects(newS3 func(context.Context, storage.S3Config) (storage.ObjectStore, error)) (storage.ObjectStore, error) { + cfg := a.applicationConfig.Distributed + if cfg.StorageURL == "" { + return nil, nil + } + if cfg.StorageBucket == "" { + return nil, fmt.Errorf("workspace S3 storage bucket must be set when storage URL is configured") + } + if (cfg.StorageAccessKey == "") != (cfg.StorageSecretKey == "") { + return nil, fmt.Errorf("workspace S3 requires both access key and secret key, or neither for the SDK credential chain") + } + if dist := a.Distributed(); dist != nil && dist.Store != nil { + return dist.Store, nil + } + store, err := newS3(a.backgroundContext(), storage.S3Config{ + Endpoint: cfg.StorageURL, Region: cfg.StorageRegion, Bucket: cfg.StorageBucket, + AccessKeyID: cfg.StorageAccessKey, SecretAccessKey: cfg.StorageSecretKey, ForcePathStyle: true, + }) + if err != nil { + return nil, fmt.Errorf("initializing workspace S3 storage: %w", err) + } + return store, nil +} diff --git a/core/application/studio_shared_files_test.go b/core/application/studio_shared_files_test.go new file mode 100644 index 000000000..57d8bf04c --- /dev/null +++ b/core/application/studio_shared_files_test.go @@ -0,0 +1,153 @@ +// SPDX-License-Identifier: MIT +package application + +import ( + "context" + "fmt" + "io" + "net/url" + "os" + "path/filepath" + "strings" + "time" + + "github.com/mudler/LocalAI/core/config" + "github.com/mudler/LocalAI/core/services/storage" + "github.com/mudler/LocalAI/core/services/studio/workspace" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "gorm.io/driver/postgres" + "gorm.io/gorm" +) + +// Embedding the interface makes every unexpected operation fail immediately; +// only calls exercised by the legacy adapter are implemented explicitly. +type sharedFilesObjectSpy struct { + storage.ObjectStore + puts int +} + +func (s *sharedFilesObjectSpy) Put(_ context.Context, _ string, r io.Reader) error { + s.puts++ + _, err := io.Copy(io.Discard, r) + return err +} +func (s *sharedFilesObjectSpy) Exists(context.Context, string) (bool, error) { return false, nil } + +var _ = Describe("PostgreSQL shared filesystem application selection", func() { + It("shares owner/frontend/worker Files automatically with independent connections without changing roles", func() { + raw := os.Getenv("LOCALAI_TEST_POSTGRES_URL") + if raw == "" { + Skip("LOCALAI_TEST_POSTGRES_URL is not set") + } + admin, err := gorm.Open(postgres.Open(raw), &gorm.Config{}) + Expect(err).NotTo(HaveOccurred()) + pool, err := admin.DB() + Expect(err).NotTo(HaveOccurred()) + schema := fmt.Sprintf("swf_app_%d", time.Now().UnixNano()) + Expect(admin.Exec("CREATE SCHEMA " + schema).Error).To(Succeed()) + DeferCleanup(func() { + Expect(admin.Exec("DROP SCHEMA " + schema + " CASCADE").Error).To(Succeed()) + Expect(pool.Close()).To(Succeed()) + }) + u, err := url.Parse(raw) + Expect(err).NotTo(HaveOccurred()) + q := u.Query() + q.Set("search_path", schema) + u.RawQuery = q.Encode() + root := filepath.Join(GinkgoT().TempDir(), "shared") + open := func(spy *sharedFilesObjectSpy) *Application { + db, err := gorm.Open(postgres.Open(u.String()), &gorm.Config{}) + Expect(err).NotTo(HaveOccurred()) + pool, err := db.DB() + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { Expect(pool.Close()).To(Succeed()) }) + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + cfg := config.NewApplicationConfig(config.WithContext(ctx), config.WithDataPath(root)) + cfg.DisableChatsIndexFiles = true + app := &Application{applicationConfig: cfg, authDB: db} + if spy != nil { + cfg.Distributed.Enabled = true + app.distributed = &DistributedServices{Store: spy} + } + Expect(app.initStudioFiles()).To(Succeed()) + Expect(app.StudioFiles().Available()).To(BeTrue()) + DeferCleanup(func() { + cancel() + if app.studioFilesTemp != "" { + Expect(os.RemoveAll(app.studioFilesTemp)).To(Succeed()) + } + }) + return app + } + owner := open(nil) + spy := &sharedFilesObjectSpy{} + frontend := open(spy) + Expect(owner.Distributed()).To(BeNil()) + Expect(owner.applicationConfig.Distributed.Enabled).To(BeFalse()) + Expect(frontend.Distributed().Store).To(BeIdenticalTo(spy)) + Expect(frontend.applicationConfig.Distributed.Enabled).To(BeTrue()) + Expect(owner.authDB).NotTo(BeIdenticalTo(frontend.authDB)) + Expect(owner.applicationConfig.DataPath).To(Equal(frontend.applicationConfig.DataPath)) + Expect(owner.StudioFiles().EnsureHome(context.Background(), "alice")).To(Succeed()) + actor := workspace.Actor{Principal: "alice"} + path, err := owner.StudioFiles().Abs("alice", "/from-owner.txt") + Expect(err).NotTo(HaveOccurred()) + _, err = owner.StudioFiles().Store().Put(context.Background(), actor, path, strings.NewReader("owner bytes"), workspace.PutOptions{}) + Expect(err).NotTo(HaveOccurred()) + read := func(app *Application) { + r, _, err := app.StudioFiles().Store().Open(context.Background(), actor, path) + Expect(err).NotTo(HaveOccurred()) + data, err := io.ReadAll(r) + Expect(err).NotTo(HaveOccurred()) + Expect(r.Close()).To(Succeed()) + Expect(string(data)).To(Equal("owner bytes")) + } + read(frontend) + worker := open(spy) + worker.applicationConfig.Distributed.ChatWorker = true + read(worker) + reopened := open(spy) + read(reopened) + Expect(spy.puts).To(BeZero(), "workspace FS must never upload through generic fallback") + }) +}) + +var _ = Describe("PostgreSQL shared filesystem startup cleanup", func() { + DescribeTable("cancels only the owned context and closes the pool on Files failure", func(early bool) { + raw := os.Getenv("LOCALAI_TEST_POSTGRES_URL") + if raw == "" { + Skip("LOCALAI_TEST_POSTGRES_URL is not set") + } + db, err := gorm.Open(postgres.Open(raw), &gorm.Config{}) + Expect(err).NotTo(HaveOccurred()) + pool, err := db.DB() + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = pool.Close() }) + caller, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + cfg := config.NewApplicationConfig(config.WithContext(caller), config.WithDataPath(GinkgoT().TempDir()), config.WithAuthEnabled(true), config.WithAuthDatabaseURL(raw)) + cleanup := workspaceStartupCleanup(cfg) + app := &Application{applicationConfig: cfg, authDB: db} + if early { + cfg.Distributed.StorageURL = "http://s3.invalid" + } else { + Expect(os.WriteFile(filepath.Join(cfg.DataPath, "chats"), []byte("blocked"), 0o600)).To(Succeed()) + } + DeferCleanup(func() { + if app.studioFilesTemp != "" { + _ = os.RemoveAll(app.studioFilesTemp) + } + }) + stopped := make(chan struct{}) + go func() { <-cfg.Context.Done(); close(stopped) }() + err = app.initStudioFiles() + Expect(app.studioFilesTemp).To(BeEmpty()) + Expect(err).To(HaveOccurred()) + cleanup(app) + Expect(pool.Ping()).To(MatchError(ContainSubstring("database is closed"))) + Eventually(stopped).Should(BeClosed()) + Expect(caller.Err()).NotTo(HaveOccurred()) + }, Entry("early S3 configuration", true), Entry("late Files init", false)) +}) diff --git a/core/application/workspace_automatic_test.go b/core/application/workspace_automatic_test.go new file mode 100644 index 000000000..c61748fa4 --- /dev/null +++ b/core/application/workspace_automatic_test.go @@ -0,0 +1,206 @@ +// SPDX-License-Identifier: MIT +package application + +import ( + "context" + "errors" + "io" + "path/filepath" + "strings" + + "github.com/mudler/LocalAI/core/config" + "github.com/mudler/LocalAI/core/services/storage" + "github.com/mudler/LocalAI/core/services/studio/files" + "github.com/mudler/LocalAI/core/services/studio/workspace" + "github.com/mudler/LocalAI/core/services/studio/workspace/blob" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +type workspaceS3Spy struct { + storage.ObjectStore + bytes map[string]string + puts int + gets int +} + +func (s *workspaceS3Spy) Put(_ context.Context, key string, r io.Reader) error { + b, err := io.ReadAll(r) + s.bytes[key] = string(b) + s.puts++ + return err +} +func (s *workspaceS3Spy) Get(_ context.Context, key string) (io.ReadCloser, error) { + s.gets++ + return io.NopCloser(strings.NewReader(s.bytes[key])), nil +} +func (s *workspaceS3Spy) Exists(_ context.Context, key string) (bool, error) { + _, ok := s.bytes[key] + return ok, nil +} + +var _ = Describe("automatic workspace", func() { + newApp := func(dp string) *Application { + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + app := &Application{applicationConfig: config.NewApplicationConfig(config.WithContext(ctx), config.WithDataPath(dp))} + app.applicationConfig.Distributed.ChatWorker = true // do not start the text-index runner in unit fixtures + DeferCleanup(func() { Expect(app.Shutdown()).To(Succeed()) }) + return app + } + withSQLite := func(app *Application) { + db, err := gorm.Open(sqlite.Open(filepath.Join(GinkgoT().TempDir(), "auth.db")), &gorm.Config{}) + Expect(err).NotTo(HaveOccurred()) + pool, err := db.DB() + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = pool.Close() }) + app.authDB = db + } + It("uses direct filesystem in distributed mode without S3 despite a generic fallback store", func() { + app := newApp(GinkgoT().TempDir()) + app.applicationConfig.Distributed.Enabled = true + spy := &sharedFilesObjectSpy{} + app.distributed = &DistributedServices{Store: spy} + Expect(app.initStudioFiles()).To(Succeed()) + Expect(app.StudioFiles().Available()).To(BeTrue()) + Expect(filepath.Join(app.applicationConfig.DataPath, "chats", "blobs")).To(BeADirectory()) + Expect(spy.puts).To(BeZero()) + }) + It("honors owner configured S3 and rejects a missing bucket rather than choosing FS", func() { + app := newApp(GinkgoT().TempDir()) + app.applicationConfig.Distributed.StorageURL = "http://s3.invalid:9000" + Expect(app.initStudioFiles()).To(MatchError(ContainSubstring("bucket"))) + Expect(app.Distributed()).To(BeNil()) + }) + DescribeTable("cleans up standalone configured S3 startup failure", func(database bool) { + caller, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + cfg := config.NewApplicationConfig(config.WithContext(caller), config.WithDataPath(GinkgoT().TempDir())) + cfg.Distributed.StorageURL = "http://s3.invalid:9000" // no bucket: fail before any network access + cfg.Auth.Enabled = database + cfg.Auth.DatabaseURL = filepath.Join(GinkgoT().TempDir(), "auth.db") + cleanup := workspaceStartupCleanup(cfg) // the same ownership scope used by New + app := &Application{applicationConfig: cfg} + DeferCleanup(func() { Expect(app.Shutdown()).To(Succeed()) }) + if database { + withSQLite(app) + } + cleaned := false + app.studioLoopsStop = func() { cleaned = true } + stopped := make(chan struct{}) + go func() { <-cfg.Context.Done(); close(stopped) }() + + Expect(app.initStudioFiles()).To(MatchError(ContainSubstring("bucket"))) + cleanup(app) + if database { + pool, err := app.authDB.DB() + Expect(err).NotTo(HaveOccurred()) + Expect(pool.Ping()).To(MatchError(ContainSubstring("database is closed"))) + } + Expect(cfg.Context.Err()).To(MatchError(context.Canceled)) + Eventually(stopped).Should(BeClosed()) + Expect(cleaned).To(BeTrue(), "startup failure must shut down already opened resources") + Expect(caller.Err()).NotTo(HaveOccurred()) + Expect(app.Distributed()).To(BeNil()) + }, Entry("SQLite owner", true), Entry("no-auth owner", false)) + DescribeTable("constructs owner S3 once and shares bytes through frontend S3 without changing global storage", func(database, noPath bool) { + dp := GinkgoT().TempDir() + if noPath { + dp = "" + } + owner := newApp(dp) + if database { + withSQLite(owner) + } + cfg := &owner.applicationConfig.Distributed + cfg.StorageURL, cfg.StorageBucket, cfg.StorageRegion = "http://s3.invalid:9000", "workspace", "test-region" + cfg.StorageAccessKey, cfg.StorageSecretKey = "test-key", "test-secret" + spy := &workspaceS3Spy{bytes: map[string]string{}} + calls := 0 + constructor := func(_ context.Context, got storage.S3Config) (storage.ObjectStore, error) { + calls++ + Expect(got).To(Equal(storage.S3Config{Endpoint: cfg.StorageURL, Bucket: cfg.StorageBucket, Region: cfg.StorageRegion, AccessKeyID: cfg.StorageAccessKey, SecretAccessKey: cfg.StorageSecretKey, ForcePathStyle: true})) + return spy, nil + } + Expect(owner.initStudioFilesWithS3(constructor)).To(Succeed()) + Expect(calls).To(Equal(1)) + Expect(owner.Distributed()).To(BeNil()) + Expect(owner.applicationConfig.Distributed.Enabled).To(BeFalse()) + if noPath { + Expect(owner.studioFilesTemp).NotTo(BeEmpty()) + } + actor := workspace.Actor{Principal: files.AdminPrincipal} + _, err := owner.StudioFiles().Store().Put(context.Background(), actor, "/shared/owner.txt", strings.NewReader("S3 owner bytes"), workspace.PutOptions{}) + Expect(err).NotTo(HaveOccurred()) + Expect(spy.puts).To(BeNumerically(">", 0)) + frontend := newApp(GinkgoT().TempDir()) + frontend.applicationConfig.Distributed = *cfg + frontend.distributed = &DistributedServices{Store: spy} + if database { + frontend.authDB = owner.authDB + } + Expect(frontend.initStudioFilesWithS3(constructor)).To(Succeed()) + Expect(calls).To(Equal(1), "must reuse distributed S3 instance") + Expect(frontend.Distributed().Store).To(BeIdenticalTo(spy)) + if database { + r, _, err := frontend.StudioFiles().Store().Open(context.Background(), actor, "/shared/owner.txt") + Expect(err).NotTo(HaveOccurred()) + b, err := io.ReadAll(r) + Expect(err).NotTo(HaveOccurred()) + Expect(r.Close()).To(Succeed()) + Expect(string(b)).To(Equal("S3 owner bytes")) + Expect(spy.gets).To(BeNumerically(">", 0)) + } + Expect(filepath.Join(frontend.applicationConfig.DataPath, "chats", "blobs")).NotTo(BeADirectory()) + }, Entry("standalone JSON", false, false), Entry("standalone SQLite", true, false), Entry("S3 cache without DataPath", true, true)) + It("validates explicit credentials, permits SDK chain, and surfaces constructor failure", func() { + app := newApp(GinkgoT().TempDir()) + cfg := &app.applicationConfig.Distributed + cfg.StorageURL, cfg.StorageBucket = "http://s3.invalid", "workspace" + calls := 0 + constructor := func(_ context.Context, got storage.S3Config) (storage.ObjectStore, error) { + calls++ + Expect(got.AccessKeyID).To(BeEmpty()) + Expect(got.SecretAccessKey).To(BeEmpty()) + return nil, errors.New("constructor failed") + } + cfg.StorageAccessKey = "key" + Expect(app.initStudioFilesWithS3(constructor)).To(MatchError(ContainSubstring("both access key"))) + cfg.StorageAccessKey, cfg.StorageSecretKey = "", "secret" + Expect(app.initStudioFilesWithS3(constructor)).To(MatchError(ContainSubstring("both access key"))) + Expect(calls).To(BeZero()) + cfg.StorageSecretKey = "" + Expect(app.initStudioFilesWithS3(constructor)).To(MatchError(ContainSubstring("constructor failed"))) + Expect(calls).To(Equal(1)) + cfg.StorageURL, cfg.StorageAccessKey = "", "unused" + Expect(app.initStudioFilesWithS3(constructor)).To(Succeed()) + Expect(calls).To(Equal(1)) + }) + It("constructs the existing S3 adapter at a fake endpoint without contacting it", func() { + app := newApp(GinkgoT().TempDir()) + cfg := &app.applicationConfig.Distributed + cfg.StorageURL, cfg.StorageBucket, cfg.StorageAccessKey, cfg.StorageSecretKey = "http://127.0.0.1:1", "workspace", "key", "secret" + objects, err := app.workspaceObjects(func(ctx context.Context, cfg storage.S3Config) (storage.ObjectStore, error) { + return storage.NewS3Store(ctx, cfg) + }) + Expect(err).NotTo(HaveOccurred()) + Expect(objects).To(BeAssignableToTypeOf(&storage.S3Store{})) + }) + It("keeps Files disabled without DataPath even with SQLite or a generic fallback", func() { + app := newApp("") + withSQLite(app) + app.distributed = &DistributedServices{Store: &sharedFilesObjectSpy{}} + Expect(app.initStudioFiles()).To(Succeed()) + Expect(app.StudioFiles().Available()).To(BeFalse()) + Expect(app.studioFilesTemp).To(BeEmpty()) + }) + It("does not treat object-store cache as a direct filesystem", func() { + fs, err := blob.NewFS(GinkgoT().TempDir()) + Expect(err).NotTo(HaveOccurred()) + Expect(fs.(interface{ IsFilesystem() bool }).IsFilesystem()).To(BeTrue()) + _, direct := blob.NewObjectStore(&sharedFilesObjectSpy{}, GinkgoT().TempDir()).(interface{ IsFilesystem() bool }) + Expect(direct).To(BeFalse()) + }) +}) diff --git a/core/cli/workspace_automatic_test.go b/core/cli/workspace_automatic_test.go new file mode 100644 index 000000000..cd006a21b --- /dev/null +++ b/core/cli/workspace_automatic_test.go @@ -0,0 +1,18 @@ +// SPDX-License-Identifier: MIT +package cli + +import ( + "github.com/alecthomas/kong" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("automatic workspace CLI", func() { + It("has no custom workspace path flag", func() { + var cmd RunCMD + parser, err := kong.New(&cmd, kong.Vars{"basepath": GinkgoT().TempDir(), "galleries": "[]", "backends": "[]", "generatedcontentpath": DefaultGeneratedContentPath(), "uploadpath": DefaultUploadPath()}) + Expect(err).NotTo(HaveOccurred()) + _, err = parser.Parse([]string{"--chats-shared-files-dir=/srv/workspace"}) + Expect(err).To(MatchError(ContainSubstring("unknown flag"))) + }) +}) diff --git a/core/config/workspace_automatic_test.go b/core/config/workspace_automatic_test.go new file mode 100644 index 000000000..14437e5af --- /dev/null +++ b/core/config/workspace_automatic_test.go @@ -0,0 +1,20 @@ +// SPDX-License-Identifier: MIT +package config_test + +import ( + "github.com/mudler/LocalAI/core/config" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "reflect" +) + +var _ = Describe("automatic workspace configuration", func() { + It("uses existing DataPath and has no additional path or coordination setting", func() { + cfg := config.NewApplicationConfig(config.WithDataPath("existing-data")) + Expect(cfg.DataPath).To(Equal("existing-data")) + _, exists := reflect.TypeOf(cfg).Elem().FieldByName("ChatsSharedFilesDir") + Expect(exists).To(BeFalse()) + _, exists = reflect.TypeOf(config.RuntimeSettings{}).FieldByName("ChatsSharedFilesDir") + Expect(exists).To(BeFalse()) + }) +}) diff --git a/core/services/studio/workspace/blob/blob.go b/core/services/studio/workspace/blob/blob.go index 3566ccd64..0f1e1b647 100644 --- a/core/services/studio/workspace/blob/blob.go +++ b/core/services/studio/workspace/blob/blob.go @@ -25,7 +25,8 @@ import ( type Store interface { // Put reads r to its end and stores the bytes. A blob that exists is not // written again. Over max bytes (when max is above zero) it stops and - // returns workspace.ErrTooLarge; an error leaves no blob. + // returns workspace.ErrTooLarge. A durability error after publication may + // leave a complete blob; callers must not publish metadata after any error. Put(ctx context.Context, r io.Reader, max int64) (sha string, size int64, err error) // Open returns a blob for reading and seeking. An unknown or malformed hash // is workspace.ErrNotFound. @@ -33,13 +34,21 @@ type Store interface { // Exists reports whether the blob is stored. Exists(ctx context.Context, sha string) (bool, error) // LinkTo makes name, a relative path in dir, a read-only file with the - // blob's bytes: a hard link when the blob is on the same file system (on - // Linux), a copy otherwise. name must not exist; its folder must. Every + // blob's bytes. Direct filesystem stores copy to isolate tool writes; + // other adapters may link protected cache files. name must not exist; its folder must. Every // step below dir goes through the root, so a link planted inside dir // cannot send the file elsewhere. An unknown hash is workspace.ErrNotFound. LinkTo(ctx context.Context, sha string, dir *os.Root, name string) error } +// Filesystem is an optional capability for authoritative direct filesystem blobs. +// Object-store caches do not implement it. PostgreSQL uses it to coordinate +// ingest/publication with purge automatically, including non-distributed owners. +type Filesystem interface { + Store + IsFilesystem() bool +} + var shaRe = regexp.MustCompile(`^[0-9a-f]{64}$`) // validSHA reports whether s is a lower-case hex SHA-256. diff --git a/core/services/studio/workspace/blob/fs.go b/core/services/studio/workspace/blob/fs.go index 0986118cd..7a6a6df7d 100644 --- a/core/services/studio/workspace/blob/fs.go +++ b/core/services/studio/workspace/blob/fs.go @@ -2,7 +2,10 @@ package blob import ( "context" + "crypto/sha256" + "encoding/hex" "errors" + "fmt" "io" "os" "path/filepath" @@ -10,75 +13,236 @@ import ( "github.com/mudler/LocalAI/core/services/studio/workspace" ) -type fsStore struct{ root string } +type fsStore struct { + root string + // fault is an instance-local test seam at publication boundaries. + fault func(string) error +} -// NewFS returns a Store that keeps blobs as read-only files under dir, in -// folders named by the first four hex digits of the hash. +// NewFS stores immutable blobs in a service-controlled directory. Shared +// filesystems must support coherent exclusive creation, rename and directory sync. func NewFS(dir string) (Store, error) { - if err := os.MkdirAll(dir, 0o700); err != nil { + abs, err := filepath.Abs(dir) + if err != nil { return nil, err } - return &fsStore{root: dir}, nil -} - -func (s *fsStore) Put(_ context.Context, r io.Reader, max int64) (string, int64, error) { - tmp, sha, size, err := ingest(r, max, filepath.Join(s.root, "tmp")) + // Preserve ordinary platform aliases (for example macOS /var). The + // operator must keep the canonical backing directory service-controlled. + if err = os.MkdirAll(abs, 0o700); err != nil { + return nil, err + } + canonical, err := filepath.EvalSymlinks(abs) if err != nil { - if tmp != "" { - _ = os.Remove(tmp) - } - return "", 0, err + return nil, err } - if err := place(tmp, shaPath(s.root, sha)); err != nil { - _ = os.Remove(tmp) - return "", 0, err - } - return sha, size, nil + return &fsStore{root: canonical}, nil } +// safe refuses symlink components inside the canonical blob root. +// The tree must remain service-controlled; arbitrary host writers are unsupported. +func (s *fsStore) safe(path string) error { + for p := path; ; p = filepath.Dir(p) { + st, err := os.Lstat(p) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + if err == nil && st.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("blob: symlink component is not allowed") + } + if p == s.root || filepath.Dir(p) == p { + break + } + } + return nil +} +func (s *fsStore) stage(name string) error { + if s.fault != nil { + return s.fault(name) + } + return nil +} +func (s *fsStore) syncDir(path string) error { + if err := s.stage("directory sync"); err != nil { + return err + } + f, err := os.Open(path) // #nosec G304 -- internal callers pass only canonical-root directories or hash shards in the service-controlled tree. + if err != nil { + return err + } + defer func() { _ = f.Close() }() + return f.Sync() +} +func (s *fsStore) mkdir(path string) error { + if err := s.safe(path); err != nil { + return err + } + if _, err := os.Stat(path); err == nil { + return nil + } else if !errors.Is(err, os.ErrNotExist) { + return err + } + if err := s.mkdir(filepath.Dir(path)); err != nil { + return err + } + if err := os.Mkdir(path, 0o700); err != nil && !errors.Is(err, os.ErrExist) { + return err + } + return s.syncDir(filepath.Dir(path)) +} +func (s *fsStore) Put(ctx context.Context, r io.Reader, max int64) (string, int64, error) { + tempDir := filepath.Join(s.root, "tmp") + if err := s.mkdir(tempDir); err != nil { + return "", 0, err + } + f, err := os.CreateTemp(tempDir, "upload-*") + if err != nil { + return "", 0, err + } + // Publication checks Sync and Close below; deferred cleanup is best-effort, + // including after a successful rename has removed the temporary name. + defer func() { + _ = f.Close() + _ = os.Remove(f.Name()) + }() + h := sha256.New() + src := io.Reader(contextReader{ctx: ctx, r: r}) + if max > 0 { + src = io.LimitReader(src, max+1) + } + if err = s.stage("write"); err != nil { + return "", 0, err + } + n, err := io.Copy(io.MultiWriter(f, h), src) + if err != nil { + return "", 0, err + } + if max > 0 && n > max { + return "", 0, workspace.ErrTooLarge + } + if err = ctx.Err(); err != nil { + return "", 0, err + } + sha := hex.EncodeToString(h.Sum(nil)) + final := shaPath(s.root, sha) + if err = s.mkdir(filepath.Dir(final)); err != nil { + return "", 0, err + } + if err = s.safe(final); err != nil { + return "", 0, err + } + if err = f.Chmod(0o400); err != nil { + return "", 0, err + } + if err = s.stage("file sync"); err != nil { + return "", 0, err + } + if err = f.Sync(); err != nil { + return "", 0, err + } + if err = f.Close(); err != nil { + return "", 0, err + } + if err = s.stage("rename"); err != nil { + return "", 0, err + } + if err = os.Rename(f.Name(), final); err != nil { + return "", 0, err + } + // Never remove final after a sync failure: another writer can own that name. + for p := filepath.Dir(final); ; p = filepath.Dir(p) { + if err = s.syncDir(p); err != nil { + return "", 0, err + } + if p == s.root { + break + } + } + if err = s.syncDir(tempDir); err != nil { + return "", 0, err + } + return sha, n, nil +} func (s *fsStore) Open(_ context.Context, sha string) (io.ReadSeekCloser, error) { if !validSHA(sha) { return nil, workspace.ErrNotFound } - f, err := os.Open(shaPath(s.root, sha)) + p := shaPath(s.root, sha) + if err := s.safe(p); err != nil { + return nil, err + } + f, err := os.Open(p) // #nosec G304 -- validSHA confines p beneath the canonical root; safe rejects in-tree symlinks in the service-controlled tree. if errors.Is(err, os.ErrNotExist) { return nil, workspace.ErrNotFound } return f, err } - -func (s *fsStore) Exists(_ context.Context, sha string) (bool, error) { - if !validSHA(sha) { +func (s *fsStore) Exists(ctx context.Context, sha string) (bool, error) { + f, err := s.Open(ctx, sha) + if errors.Is(err, workspace.ErrNotFound) { return false, nil } - _, err := os.Stat(shaPath(s.root, sha)) - if errors.Is(err, os.ErrNotExist) { - return false, nil + if err != nil { + return false, err } - return err == nil, err + return true, f.Close() } -func (s *fsStore) LinkTo(_ context.Context, sha string, dir *os.Root, name string) error { - if !validSHA(sha) { - return workspace.ErrNotFound +// LinkTo deliberately copies. The same UID can chmod a read-only hard link and +// mutate its inode, so a hard link is not an isolation boundary for tool files. +func (s *fsStore) LinkTo(ctx context.Context, sha string, dir *os.Root, name string) error { + if !filepath.IsLocal(name) { + return workspace.ErrInvalid } - err := linkOrCopy(shaPath(s.root, sha), dir, name) - if errors.Is(err, os.ErrNotExist) { - if _, serr := os.Stat(shaPath(s.root, sha)); serr != nil { - return workspace.ErrNotFound - } + in, err := s.Open(ctx, sha) + if err != nil { + return err + } + defer func() { _ = in.Close() }() + out, err := dir.OpenFile(name, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o400) + if err != nil { + return err + } + _, err = io.Copy(out, in) + closeErr := out.Close() + if err == nil { + err = closeErr + } + if err != nil { + _ = dir.Remove(name) } return err } - -// Remove deletes the blob file. The file may be read-only; its folder is not. func (s *fsStore) Remove(_ context.Context, sha string) error { if !validSHA(sha) { return nil } - err := os.Remove(shaPath(s.root, sha)) + p := shaPath(s.root, sha) + if err := s.safe(p); err != nil { + return err + } + err := os.Remove(p) if errors.Is(err, os.ErrNotExist) { return nil } - return err + if err != nil { + return err + } + return s.syncDir(filepath.Dir(p)) } + +// A Reader blocked inside Read must honor its own request cancellation. +// Between reads we stop promptly, including before publication after EOF. +type contextReader struct { + ctx context.Context + r io.Reader +} + +func (r contextReader) Read(p []byte) (int, error) { + if err := r.ctx.Err(); err != nil { + return 0, err + } + return r.r.Read(p) +} + +// IsFilesystem distinguishes authoritative direct FS from an object-store cache. +func (*fsStore) IsFilesystem() bool { return true } diff --git a/core/services/studio/workspace/blob/publication_internal_test.go b/core/services/studio/workspace/blob/publication_internal_test.go new file mode 100644 index 000000000..6300c3ca4 --- /dev/null +++ b/core/services/studio/workspace/blob/publication_internal_test.go @@ -0,0 +1,118 @@ +// SPDX-License-Identifier: MIT +package blob + +import ( + "context" + "errors" + "io" + "os" + "path/filepath" + "strings" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("shared filesystem publication", func() { + for _, stage := range []string{"write", "file sync", "rename", "directory sync"} { + It("cleans up after injected "+stage+" failure", func() { + root := GinkgoT().TempDir() + st, err := NewFS(root) + Expect(err).NotTo(HaveOccurred()) + s := st.(*fsStore) + content := "complete" + seed, err := NewFS(GinkgoT().TempDir()) + Expect(err).NotTo(HaveOccurred()) + sha, _, err := seed.Put(context.Background(), strings.NewReader(content), 0) + Expect(err).NotTo(HaveOccurred()) + // Pre-create shard/tmp so directory-sync injection tests post-publication. + Expect(os.MkdirAll(filepath.Dir(shaPath(root, sha)), 0o700)).To(Succeed()) + Expect(os.Mkdir(filepath.Join(root, "tmp"), 0o700)).To(Succeed()) + injected := errors.New("injected " + stage) + s.fault = func(at string) error { + if at == stage { + return injected + } + return nil + } + _, _, err = s.Put(context.Background(), strings.NewReader(content), 0) + Expect(err).To(MatchError(injected)) + entries, err := os.ReadDir(filepath.Join(root, "tmp")) + Expect(err).NotTo(HaveOccurred()) + Expect(entries).To(BeEmpty()) + s.fault = nil + f, err := s.Open(context.Background(), sha) + if stage == "directory sync" { + Expect(err).NotTo(HaveOccurred()) + defer func() { _ = f.Close() }() + got, err := io.ReadAll(f) + Expect(err).NotTo(HaveOccurred()) + Expect(string(got)).To(Equal(content)) + } else { + if err == nil { + _ = f.Close() + } + Expect(err).To(HaveOccurred(), "failed upload must not be published") + } + }) + } + + It("syncs newly created ancestors", func() { + st, err := NewFS(GinkgoT().TempDir()) + Expect(err).NotTo(HaveOccurred()) + s := st.(*fsStore) + count := 0 + s.fault = func(at string) error { + if at == "directory sync" { + count++ + } + return nil + } + _, _, err = s.Put(context.Background(), strings.NewReader("new shards"), 0) + Expect(err).NotTo(HaveOccurred()) + // Creation of tmp/two shard parents, final/ancestor directories and tmp. + Expect(count).To(Equal(7)) + }) + + It("preserves concurrent publication after a failed sync", func() { + root := GinkgoT().TempDir() + first, err := NewFS(root) + Expect(err).NotTo(HaveOccurred()) + other, err := NewFS(root) + Expect(err).NotTo(HaveOccurred()) + content := "concurrent publication" + sha, _, err := other.Put(context.Background(), strings.NewReader(content), 0) + Expect(err).NotTo(HaveOccurred()) + injected := errors.New("post-rename sync") + first.(*fsStore).fault = func(at string) error { + if at != "directory sync" { + return nil + } + // Another instance publishes the same hash after this writer's rename. + _, _, err := other.Put(context.Background(), strings.NewReader(content), 0) + Expect(err).NotTo(HaveOccurred()) + return injected + } + _, _, err = first.Put(context.Background(), strings.NewReader(content), 0) + Expect(err).To(MatchError(injected)) + f, err := other.Open(context.Background(), sha) + Expect(err).NotTo(HaveOccurred()) + defer func() { _ = f.Close() }() + got, err := io.ReadAll(f) + Expect(err).NotTo(HaveOccurred()) + Expect(string(got)).To(Equal(content)) + }) + + It("honors cancellation at the end of bounded ingest", func() { + s, err := NewFS(GinkgoT().TempDir()) + Expect(err).NotTo(HaveOccurred()) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + _, _, err = s.Put(ctx, cancelAtEOF{cancel}, 128) + Expect(err).To(MatchError(context.Canceled)) + }) +}) + +type cancelAtEOF struct{ cancel context.CancelFunc } + +func (r cancelAtEOF) Read([]byte) (int, error) { r.cancel(); return 0, io.EOF } diff --git a/core/services/studio/workspace/blob/shared_fs_test.go b/core/services/studio/workspace/blob/shared_fs_test.go new file mode 100644 index 000000000..ffa4b3a90 --- /dev/null +++ b/core/services/studio/workspace/blob/shared_fs_test.go @@ -0,0 +1,234 @@ +// SPDX-License-Identifier: MIT +package blob_test + +import ( + "bufio" + "context" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/core/services/studio/workspace/blob" +) + +var _ = Describe("shared filesystem", func() { + ctx := context.Background() + It("does not let a materialized file mutate the authoritative blob", func() { + s, err := blob.NewFS(GinkgoT().TempDir()) + Expect(err).NotTo(HaveOccurred()) + sha, _, err := s.Put(ctx, strings.NewReader("immutable"), 0) + Expect(err).NotTo(HaveOccurred()) + dest := GinkgoT().TempDir() + root, err := os.OpenRoot(dest) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = root.Close() }) + Expect(s.LinkTo(ctx, sha, root, "tool-file")).To(Succeed()) + // The service UID owns the materialized inode and can chmod it. Mode 0444 + // alone cannot isolate an authoritative inode from a tool's writes. + Expect(os.Chmod(filepath.Join(dest, "tool-file"), 0o600)).To(Succeed()) + Expect(os.WriteFile(filepath.Join(dest, "tool-file"), []byte("changed"), 0o600)).To(Succeed()) + r, err := s.Open(ctx, sha) + Expect(err).NotTo(HaveOccurred()) + defer func() { _ = r.Close() }() + got, err := io.ReadAll(r) + Expect(err).NotTo(HaveOccurred()) + Expect(string(got)).To(Equal("immutable")) + }) + It("rejects a symlinked temporary directory without writing outside the root", func() { + dir, outside := GinkgoT().TempDir(), GinkgoT().TempDir() + s, err := blob.NewFS(dir) + Expect(err).NotTo(HaveOccurred()) + Expect(os.Symlink(outside, filepath.Join(dir, "tmp"))).To(Succeed()) + _, _, err = s.Put(ctx, strings.NewReader("private upload"), 0) + Expect(err).To(HaveOccurred()) + entries, err := os.ReadDir(outside) + Expect(err).NotTo(HaveOccurred()) + Expect(entries).To(BeEmpty()) + }) + It("rejects symlinked shard and final paths and keeps restrictive permissions", func() { + for _, where := range []string{"shard", "final"} { + root, outside := GinkgoT().TempDir(), GinkgoT().TempDir() + s, err := blob.NewFS(root) + Expect(err).NotTo(HaveOccurred()) + sha := sum("content") + shard := filepath.Join(root, sha[:2], sha[2:4]) + if where == "shard" { + Expect(os.Symlink(outside, filepath.Join(root, sha[:2]))).To(Succeed()) + } else { + Expect(os.MkdirAll(shard, 0o700)).To(Succeed()) + Expect(os.WriteFile(filepath.Join(outside, "secret"), []byte("secret"), 0o600)).To(Succeed()) + Expect(os.Symlink(filepath.Join(outside, "secret"), filepath.Join(shard, sha))).To(Succeed()) + } + _, _, err = s.Put(ctx, strings.NewReader("content"), 0) + Expect(err).To(HaveOccurred()) + _, err = s.Open(ctx, sha) + Expect(err).To(HaveOccurred()) + } + root := GinkgoT().TempDir() + s, err := blob.NewFS(root) + Expect(err).NotTo(HaveOccurred()) + sha, _, err := s.Put(ctx, strings.NewReader("content"), 0) + Expect(err).NotTo(HaveOccurred()) + for _, path := range []string{filepath.Join(root, "tmp"), filepath.Join(root, sha[:2]), filepath.Join(root, sha[:2], sha[2:4])} { + st, err := os.Stat(path) + Expect(err).NotTo(HaveOccurred()) + Expect(st.Mode().Perm() & 0o077).To(BeZero()) + } + st, err := os.Stat(filepath.Join(root, sha[:2], sha[2:4], sha)) + Expect(err).NotTo(HaveOccurred()) + Expect(st.Mode().Perm() & 0o222).To(BeZero()) + }) + + It("preserves default filesystem roots beneath platform directory aliases", func() { + parent := GinkgoT().TempDir() + real := filepath.Join(parent, "real") + alias := filepath.Join(parent, "alias") + Expect(os.Mkdir(real, 0o700)).To(Succeed()) + Expect(os.Symlink(real, alias)).To(Succeed()) + s, err := blob.NewFS(filepath.Join(alias, "blobs")) + Expect(err).NotTo(HaveOccurred()) + sha, _, err := s.Put(ctx, strings.NewReader("platform alias"), 1<<20) + Expect(err).NotTo(HaveOccurred()) + r, err := s.Open(ctx, sha) + Expect(err).NotTo(HaveOccurred()) + defer func() { _ = r.Close() }() + got, err := io.ReadAll(r) + Expect(err).NotTo(HaveOccurred()) + Expect(string(got)).To(Equal("platform alias")) + }) + +}) + +// Re-exec the already compiled binary; child errors propagate through Wait. +func TestSharedFSProcess(t *testing.T) { + mode := os.Getenv("SWF_CHILD") + if mode == "" { + return + } + // This bootstrap runs without the parent Ginkgo suite in each OS process. + g := NewWithT(t) + s, err := blob.NewFS(os.Getenv("SWF_ROOT")) + g.Expect(err).NotTo(HaveOccurred()) + content := os.Getenv("SWF_CONTENT") + if mode == "read" { + r, err := s.Open(context.Background(), sum(content)) + g.Expect(err).NotTo(HaveOccurred()) + _, err = r.Seek(1, io.SeekStart) + g.Expect(err).NotTo(HaveOccurred()) + got, err := io.ReadAll(r) + g.Expect(r.Close()).To(Succeed()) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(string(got)).To(Equal(content[1:])) + root, err := os.OpenRoot(os.Getenv("SWF_DEST")) + g.Expect(err).NotTo(HaveOccurred()) + defer func() { _ = root.Close() }() + g.Expect(s.LinkTo(context.Background(), sum(content), root, "file")).To(Succeed()) + return + } + fmt.Println("ready") + _, err = bufio.NewReader(os.Stdin).ReadByte() + g.Expect(err).NotTo(HaveOccurred()) + var reader io.Reader = strings.NewReader(content) + if mode == "interrupt" { + reader = &interruptedUpload{content: content} + } + _, _, err = s.Put(context.Background(), reader, 0) + g.Expect(err).NotTo(HaveOccurred()) +} + +type interruptedUpload struct { + content string + sent bool +} + +func (r *interruptedUpload) Read(p []byte) (int, error) { + if !r.sent { + r.sent = true + return copy(p, r.content[:len(r.content)/2]), nil + } + fmt.Println("partial") + _, err := bufio.NewReader(os.Stdin).ReadByte() + return 0, err +} + +var _ = Describe("shared filesystem independent processes", func() { + It("publishes concurrent identical and distinct contents and reads after restart", func() { + root := GinkgoT().TempDir() + type child struct { + cmd *exec.Cmd + in io.WriteCloser + } + var children []child + for _, content := range []string{"identical bytes", "identical bytes", "distinct bytes"} { + cmd := exec.Command(os.Args[0], "-test.run=^TestSharedFSProcess$") + cmd.Env = append(os.Environ(), "SWF_CHILD=put", "SWF_ROOT="+root, "SWF_CONTENT="+content) + in, err := cmd.StdinPipe() + Expect(err).NotTo(HaveOccurred()) + out, err := cmd.StdoutPipe() + Expect(err).NotTo(HaveOccurred()) + cmd.Stderr = GinkgoWriter + Expect(cmd.Start()).To(Succeed()) + line, err := bufio.NewReader(out).ReadString('\n') + Expect(err).NotTo(HaveOccurred()) + Expect(line).To(Equal("ready\n")) + children = append(children, child{cmd, in}) + } + for _, c := range children { + _, err := c.in.Write([]byte("x")) + Expect(err).NotTo(HaveOccurred()) + Expect(c.in.Close()).To(Succeed()) + } + for _, c := range children { + Expect(c.cmd.Wait()).To(Succeed()) + } + for _, content := range []string{"identical bytes", "distinct bytes"} { + dest := GinkgoT().TempDir() + cmd := exec.Command(os.Args[0], "-test.run=^TestSharedFSProcess$") + cmd.Env = append(os.Environ(), "SWF_CHILD=read", "SWF_ROOT="+root, "SWF_CONTENT="+content, "SWF_DEST="+dest) + out, err := cmd.CombinedOutput() + Expect(err).NotTo(HaveOccurred(), string(out)) + got, err := os.ReadFile(filepath.Join(dest, "file")) + Expect(err).NotTo(HaveOccurred()) + Expect(string(got)).To(Equal(content)) + } + }) + It("never publishes a partial hash when an uploader dies after partial ingest", func() { + root := GinkgoT().TempDir() + content := strings.Repeat("incomplete", 1000) + cmd := exec.Command(os.Args[0], "-test.run=^TestSharedFSProcess$") + cmd.Env = append(os.Environ(), "SWF_CHILD=interrupt", "SWF_ROOT="+root, "SWF_CONTENT="+content) + in, err := cmd.StdinPipe() + Expect(err).NotTo(HaveOccurred()) + defer func() { _ = in.Close() }() + out, err := cmd.StdoutPipe() + Expect(err).NotTo(HaveOccurred()) + Expect(cmd.Start()).To(Succeed()) + lines := bufio.NewReader(out) + line, err := lines.ReadString('\n') + Expect(err).NotTo(HaveOccurred()) + Expect(line).To(Equal("ready\n")) + _, err = in.Write([]byte("x")) + Expect(err).NotTo(HaveOccurred()) + line, err = lines.ReadString('\n') + Expect(err).NotTo(HaveOccurred()) + Expect(line).To(Equal("partial\n")) + Expect(cmd.Process.Kill()).To(Succeed()) + Expect(cmd.Wait()).NotTo(Succeed()) + s, err := blob.NewFS(root) + Expect(err).NotTo(HaveOccurred()) + exists, err := s.Exists(context.Background(), sum(content)) + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeFalse()) + files, err := os.ReadDir(root) + Expect(err).NotTo(HaveOccurred()) + Expect(files).To(HaveLen(1)) + Expect(files[0].Name()).To(Equal("tmp")) + }) +}) diff --git a/core/services/studio/workspace/gormstore/files.go b/core/services/studio/workspace/gormstore/files.go index 92e02f71f..e7653b2bf 100644 --- a/core/services/studio/workspace/gormstore/files.go +++ b/core/services/studio/workspace/gormstore/files.go @@ -41,19 +41,29 @@ func (s *Workspace) Put(ctx context.Context, a Actor, p string, r io.Reader, o P if err := s.precheckPut(ctx, s.db, a, segs); err != nil { return Node{}, err } - sum, size, err := s.blobs.Put(ctx, r, s.limit(o.MaxSize)) - if err != nil { - return Node{}, err - } title := "Upload " + segs[len(segs)-1] if o.Title != "" { title = o.Title } - w, err := s.beginWrite(ctx, a, title, ChangeUpload) + var w *writer + if s.sharedFilesystem { + w, err = s.beginWrite(ctx, a, title, ChangeUpload) + if err != nil { + return Node{}, err + } + defer w.rollback() + } + sum, size, err := s.blobs.Put(ctx, r, s.limit(o.MaxSize)) if err != nil { return Node{}, err } - defer w.rollback() + if w == nil { + w, err = s.beginWrite(ctx, a, title, ChangeUpload) + if err != nil { + return Node{}, err + } + defer w.rollback() + } n, asked, err := w.put(segs, sum, size, o) if err != nil { return Node{}, err diff --git a/core/services/studio/workspace/gormstore/gormstore_postgres_test.go b/core/services/studio/workspace/gormstore/gormstore_postgres_test.go index af6bea17e..1778fd2ef 100644 --- a/core/services/studio/workspace/gormstore/gormstore_postgres_test.go +++ b/core/services/studio/workspace/gormstore/gormstore_postgres_test.go @@ -2,6 +2,7 @@ package gormstore_test import ( "fmt" + "net/url" "os" "path/filepath" "sync/atomic" @@ -44,7 +45,9 @@ var _ = Describe("The PostgreSQL store", func() { _ = admin.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`).Error _ = adminRaw.Close() }) - db, err := gorm.Open(postgres.Open(url+"&search_path="+schema), &gorm.Config{}) + dsn, err := urlpkgParse(url, schema) + Expect(err).NotTo(HaveOccurred()) + db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{}) Expect(err).ToNot(HaveOccurred()) raw, _ := db.DB() DeferCleanup(func() { _ = raw.Close() }) @@ -56,3 +59,14 @@ var _ = Describe("The PostgreSQL store", func() { }) }) }) + +func urlpkgParse(raw, schema string) (string, error) { + u, err := url.Parse(raw) + if err != nil { + return "", err + } + q := u.Query() + q.Set("search_path", schema) + u.RawQuery = q.Encode() + return u.String(), nil +} diff --git a/core/services/studio/workspace/gormstore/purge.go b/core/services/studio/workspace/gormstore/purge.go index b5d708bb3..f988fd77b 100644 --- a/core/services/studio/workspace/gormstore/purge.go +++ b/core/services/studio/workspace/gormstore/purge.go @@ -6,8 +6,10 @@ import ( "context" "database/sql" "errors" + "github.com/mudler/xlog" "slices" "strings" + "time" "github.com/mudler/LocalAI/core/services/studio/workspace/blob" ) @@ -249,18 +251,25 @@ func (s *Workspace) purge(ctx context.Context, a Actor, dryRun bool, pick func(q for sha := range freed { shas = append(shas, sha) } - blob.RemoveLater(s.blobs, shas, blob.Grace(s.grace), s.blobsInUse) + if s.sharedFilesystem { + s.removePGBlobs(ctx, shas) + } else { + blob.RemoveLater(s.blobs, shas, blob.Grace(s.grace), s.blobsInUse) + } return res, nil } // blobsInUse tells which of shas a node or a journal row refers to. func (s *Workspace) blobsInUse(ctx context.Context, shas []string) (map[string]bool, error) { + return blobsInUse(ctx, s.db, shas) +} +func blobsInUse(ctx context.Context, q querier, shas []string) (map[string]bool, error) { used := make(map[string]bool, len(shas)) err := chunks(shas, func(part []string) error { list := inList(len(part)) args := idArgs(part) all := append(append(append([]any{}, args...), args...), args...) - rows, err := s.db.QueryContext(ctx, `SELECT blob FROM studio_ws_nodes WHERE blob IN (`+list+`) + rows, err := q.QueryContext(ctx, `SELECT blob FROM studio_ws_nodes WHERE blob IN (`+list+`) UNION SELECT old_blob FROM studio_ws_journal WHERE old_blob IN (`+list+`) UNION SELECT new_blob FROM studio_ws_journal WHERE new_blob IN (`+list+`)`, all...) if err != nil { @@ -282,3 +291,53 @@ func scanStrings(rows *sql.Rows, into map[string]bool) error { } return rows.Err() } + +// The revision row covers ingest through reference publication and this entire +// check/delete interval across PostgreSQL connections. Grace alone cannot do so. +func (s *Workspace) removePGBlobs(ctx context.Context, shas []string) { + rm, ok := s.blobs.(blob.Remover) + if !ok || len(shas) == 0 { + return + } + if grace := blob.Grace(s.grace); grace > 0 { + // Delayed work outlives the request; its timeout starts after grace. + time.AfterFunc(grace, func() { s.removePGBlobsNow(context.Background(), shas, rm) }) + } else { + s.removePGBlobsNow(ctx, shas, rm) + } +} + +// removePGBlobsNow keeps the check/delete transaction in one cancellation scope. +func (s *Workspace) removePGBlobsNow(caller context.Context, shas []string, rm blob.Remover) { + // Bound pool acquisition, revision-lock waiting and reference checks together. + ctx, cancel := context.WithTimeout(caller, 30*time.Second) + defer cancel() + t, err := s.db.BeginTx(ctx) + if err != nil { + xlog.Warn("workspace: could not begin blob purge", "error", err) + return + } + defer func() { _ = t.Rollback() }() + if _, err = t.ExecContext(ctx, `UPDATE studio_ws_meta SET value = value WHERE key = 'rev'`); err != nil { + xlog.Warn("workspace: could not lock blob purge", "error", err) + return + } + used, err := blobsInUse(ctx, t, shas) + if err != nil { + xlog.Warn("workspace: could not check the blobs of a purge", "error", err) + return + } + for _, sha := range shas { + if ctx.Err() != nil { + return + } + if !used[sha] { + if err := rm.Remove(ctx, sha); err != nil { + xlog.Warn("workspace: could not remove a purged blob", "error", err) + } + } + } + if err := t.Commit(); err != nil { + xlog.Warn("workspace: could not finish blob purge", "error", err) + } +} diff --git a/core/services/studio/workspace/gormstore/shared_fs_postgres_test.go b/core/services/studio/workspace/gormstore/shared_fs_postgres_test.go new file mode 100644 index 000000000..6ccd8c1f5 --- /dev/null +++ b/core/services/studio/workspace/gormstore/shared_fs_postgres_test.go @@ -0,0 +1,344 @@ +// SPDX-License-Identifier: MIT +package gormstore + +import ( + "context" + "database/sql" + "fmt" + "io" + "net/url" + "os" + "strings" + "sync" + "sync/atomic" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "gorm.io/driver/postgres" + "gorm.io/gorm" + + "github.com/mudler/LocalAI/core/services/studio/workspace" + "github.com/mudler/LocalAI/core/services/studio/workspace/blob" + "github.com/mudler/LocalAI/core/services/studio/workspace/wstest" +) + +var sharedSchemaSeq atomic.Int64 + +// gateBlob exposes explicit ingest and removal barriers without timing guesses. +type gateBlob struct { + blob.Store + pool *sql.DB + mu sync.Mutex + uploaded, resume chan struct{} + deleting, removeResume chan struct{} +} + +func (b *gateBlob) IsFilesystem() bool { + fs, ok := b.Store.(blob.Filesystem) + return ok && fs.IsFilesystem() +} + +func (b *gateBlob) Put(ctx context.Context, r io.Reader, max int64) (string, int64, error) { + sha, n, err := b.Store.Put(ctx, r, max) + b.mu.Lock() + entered, resume := b.uploaded, b.resume + b.mu.Unlock() + if err == nil && entered != nil { + close(entered) + <-resume + } + return sha, n, err +} +func (b *gateBlob) Remove(ctx context.Context, sha string) error { + b.mu.Lock() + entered, resume := b.deleting, b.removeResume + b.mu.Unlock() + if entered != nil { + close(entered) + <-resume + } + return b.Store.(blob.Remover).Remove(ctx, sha) +} + +func sharedPG() (*Workspace, *Workspace, *gateBlob, *gateBlob) { + GinkgoHelper() + rawURL := os.Getenv("LOCALAI_TEST_POSTGRES_URL") + if rawURL == "" { + Skip("LOCALAI_TEST_POSTGRES_URL is not set") + } + u, err := url.Parse(rawURL) + Expect(err).NotTo(HaveOccurred()) + schema := fmt.Sprintf("swf_%d_%d", os.Getpid(), sharedSchemaSeq.Add(1)) + admin, err := gorm.Open(postgres.Open(rawURL), &gorm.Config{}) + Expect(err).NotTo(HaveOccurred()) + adminRaw, err := admin.DB() + Expect(err).NotTo(HaveOccurred()) + Expect(admin.Exec("CREATE SCHEMA " + schema).Error).To(Succeed()) + DeferCleanup(func() { + Expect(admin.Exec("DROP SCHEMA " + schema + " CASCADE").Error).To(Succeed()) + Expect(adminRaw.Close()).To(Succeed()) + }) + q := u.Query() + q.Set("search_path", schema) + u.RawQuery = q.Encode() + root := GinkgoT().TempDir() + open := func() (*Workspace, *gateBlob) { + db, err := gorm.Open(postgres.Open(u.String()), &gorm.Config{}) + Expect(err).NotTo(HaveOccurred()) + pool, err := db.DB() + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { Expect(pool.Close()).To(Succeed()) }) + fs, err := blob.NewFS(root) + Expect(err).NotTo(HaveOccurred()) + g := &gateBlob{Store: fs, pool: pool} + s, err := New(db, g, Options{Admin: wstest.Admin, BlobGrace: -1}) + Expect(err).NotTo(HaveOccurred()) + return s, g + } + a, ba := open() + b, bb := open() + return a, b, ba, bb +} + +var _ = Describe("PostgreSQL shared filesystem", func() { + ctx := context.Background() + actor := wstest.AdminActor + It("bounds cleanup waiting on the revision row and preserves blobs", func() { + _, b, ba, bb := sharedPG() + sha, _, err := bb.Store.Put(ctx, strings.NewReader("unreferenced"), 0) + Expect(err).NotTo(HaveOccurred()) + lock, err := ba.pool.BeginTx(ctx, nil) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = lock.Rollback() }) + _, err = lock.ExecContext(ctx, `UPDATE studio_ws_meta SET value = value WHERE key = 'rev'`) + Expect(err).NotTo(HaveOccurred()) + cleanup, cancel := context.WithTimeout(ctx, 500*time.Millisecond) + defer cancel() + done := make(chan struct{}) + go func() { b.removePGBlobsNow(cleanup, []string{sha}, bb); close(done) }() + waitSharedLock(ba.pool) + Eventually(done, 2*time.Second).Should(BeClosed()) + Eventually(func() int { return bb.pool.Stats().InUse }, time.Second).Should(BeZero()) + exists, err := bb.Store.Exists(ctx, sha) + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeTrue()) + }) + + It("honors immediate caller cancellation without deleting blobs", func() { + _, b, _, bb := sharedPG() + sha, _, err := bb.Store.Put(ctx, strings.NewReader("canceled cleanup"), 0) + Expect(err).NotTo(HaveOccurred()) + caller, cancel := context.WithCancel(ctx) + cancel() + b.removePGBlobs(caller, []string{sha}) + exists, err := bb.Store.Exists(ctx, sha) + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeTrue()) + Expect(bb.pool.Stats().InUse).To(BeZero()) + }) + + It("lets delayed cleanup outlive caller cancellation", func() { + _, b, _, bb := sharedPG() + sha, _, err := bb.Store.Put(ctx, strings.NewReader("delayed cleanup"), 0) + Expect(err).NotTo(HaveOccurred()) + b.grace = 20 * time.Millisecond + caller, cancel := context.WithCancel(ctx) + cancel() + b.removePGBlobs(caller, []string{sha}) + Eventually(func() bool { + exists, err := bb.Store.Exists(ctx, sha) + Expect(err).NotTo(HaveOccurred()) + return exists + }, 2*time.Second).Should(BeFalse()) + Eventually(func() int { return bb.pool.Stats().InUse }, time.Second).Should(BeZero()) + }) + + It("shares owner writes, authorization and materialization across independent pools", func() { + a, b, _, _ := sharedPG() + wstest.MustPut(a, actor, "/owner/file", "shared") + Expect(wstest.MustRead(b, actor, "/owner/file")).To(Equal("shared")) + _, _, err := b.Open(ctx, wstest.Bob, "/owner/file") + Expect(err).To(MatchError(workspace.ErrNotFound)) + v := workspace.View{Actor: actor, Mounts: []workspace.Mount{{NodeID: 1, At: "files", Mode: workspace.ModeWrite}}} + dir := GinkgoT().TempDir() + _, err = workspace.MaterializeAt(ctx, b, v, dir) + Expect(err).NotTo(HaveOccurred()) + got, err := os.ReadFile(dir + "/files/owner/file") + Expect(err).NotTo(HaveOccurred()) + Expect(string(got)).To(Equal("shared")) + }) + for _, kind := range []string{"Put", "CommitUpper"} { + It("keeps a newly referenced blob when a purge overlaps "+kind+" before metadata", func() { + a, b, ba, _ := sharedPG() + wstest.MustPut(a, actor, "/old", "same") + Expect(a.Delete(ctx, actor, "/old")).To(Succeed()) + ba.uploaded = make(chan struct{}) + ba.resume = make(chan struct{}) + done := make(chan error, 1) + go func() { + var err error + if kind == "Put" { + _, err = a.Put(ctx, actor, "/new", strings.NewReader("same"), workspace.PutOptions{}) + } else { + v := workspace.View{Actor: actor, Mounts: []workspace.Mount{{NodeID: 1, At: "files", Mode: workspace.ModeWrite}}} + _, _, err = a.CommitUpper(ctx, v, actor, 0, workspace.FixtureSource{workspace.FixtureFile("files/new", "same")}, workspace.UpperOptions{}) + } + done <- err + }() + <-ba.uploaded + purged := make(chan error, 1) + go func() { _, err := b.PurgeAll(ctx, actor, "/", false); purged <- err }() + waitSharedLock(ba.pool) + close(ba.resume) + Expect(<-purged).To(Succeed()) + Expect(<-done).To(Succeed()) + Expect(wstest.MustRead(b, actor, "/new")).To(Equal("same")) + }) + } + for _, kind := range []string{"Put", "CommitUpper", "transaction"} { + It("serializes reference-check/delete with "+kind+" publication", func() { + a, b, _, bb := sharedPG() + wstest.MustPut(a, actor, "/old", "same") + Expect(a.Delete(ctx, actor, "/old")).To(Succeed()) + bb.deleting = make(chan struct{}) + bb.removeResume = make(chan struct{}) + purged := make(chan error, 1) + go func() { _, err := b.PurgeAll(ctx, actor, "/", false); purged <- err }() + <-bb.deleting // Reference check finished; deletion has not happened. + written := make(chan error, 1) + go func() { + var err error + switch kind { + case "Put": + _, err = a.Put(ctx, actor, "/new", strings.NewReader("same"), workspace.PutOptions{}) + case "transaction": + var tx workspace.Tx + tx, err = a.Begin(ctx, actor, "race") + if err == nil { + _, err = tx.Put(ctx, "/new", strings.NewReader("same"), workspace.PutOptions{}) + if err == nil { + _, err = tx.Commit() + } else { + _ = tx.Rollback() + } + } + case "CommitUpper": + v := workspace.View{Actor: actor, Mounts: []workspace.Mount{{NodeID: 1, At: "files", Mode: workspace.ModeWrite}}} + _, _, err = a.CommitUpper(ctx, v, actor, 0, workspace.FixtureSource{workspace.FixtureFile("files/new", "same")}, workspace.UpperOptions{}) + } + written <- err + }() + waitSharedLock(bb.pool) + close(bb.removeResume) + Expect(<-purged).To(Succeed()) + Expect(<-written).To(Succeed()) + Expect(wstest.MustRead(b, actor, "/new")).To(Equal("same")) + }) + } + It("keeps retained journal references and permits purge/re-upload", func() { + a, b, _, _ := sharedPG() + wstest.MustPut(a, actor, "/history", "old") + wstest.MustPut(a, actor, "/history", "current") + wstest.MustPut(a, actor, "/deleted", "old") + Expect(a.Delete(ctx, actor, "/deleted")).To(Succeed()) + _, err := b.PurgeAll(ctx, actor, "/", false) + Expect(err).NotTo(HaveOccurred()) + // Undo needs the retained old_blob even though no live node names it. + history, err := a.History(ctx, actor, workspace.HistoryQuery{Path: "/history"}) + Expect(err).NotTo(HaveOccurred()) + _, _, err = b.Undo(ctx, actor, history[0].Rev, workspace.UndoOptions{}) + Expect(err).NotTo(HaveOccurred()) + Expect(wstest.MustRead(a, actor, "/history")).To(Equal("old")) + wstest.MustPut(a, actor, "/deleted", "old") + Expect(wstest.MustRead(b, actor, "/deleted")).To(Equal("old")) + }) + It("shares concurrent same and different content publications", func() { + a, b, _, _ := sharedPG() + done := make(chan error, 3) + for i, content := range []string{"same", "same", "different"} { + s := a + if i%2 == 1 { + s = b + } + path := fmt.Sprintf("/concurrent/%d", i) + go func() { + _, err := s.Put(ctx, actor, path, strings.NewReader(content), workspace.PutOptions{}) + done <- err + }() + } + for i := 0; i < 3; i++ { + Expect(<-done).To(Succeed()) + } + for i, content := range []string{"same", "same", "different"} { + Expect(wstest.MustRead(b, actor, fmt.Sprintf("/concurrent/%d", i))).To(Equal(content)) + } + }) + + It("cancels slow shared ingest and releases the revision lock", func() { + a, b, ba, _ := sharedPG() + request, cancel := context.WithCancel(ctx) + defer cancel() + reader := &cancelUpload{ctx: request, entered: make(chan struct{})} + done := make(chan error, 1) + go func() { _, err := a.Put(request, actor, "/slow", reader, workspace.PutOptions{}); done <- err }() + <-reader.entered + waiter := make(chan error, 1) + go func() { _, err := b.Mkdir(ctx, actor, "/after"); waiter <- err }() + waitSharedLock(ba.pool) + cancel() + Expect(<-done).To(MatchError(context.Canceled)) + Expect(<-waiter).To(Succeed()) + _, err := b.Stat(ctx, actor, "/slow") + Expect(err).To(MatchError(workspace.ErrNotFound)) + }) + + for _, kind := range []string{"Put", "CommitUpper"} { + It("automatically coordinates default PostgreSQL direct-FS ingest for "+kind, func() { + a, b, ba, _ := sharedPG() + ba.uploaded = make(chan struct{}) + ba.resume = make(chan struct{}) + done := make(chan error, 1) + go func() { + var err error + if kind == "Put" { + _, err = a.Put(ctx, actor, "/new", strings.NewReader("content"), workspace.PutOptions{}) + } else { + v := workspace.View{Actor: actor, Mounts: []workspace.Mount{{NodeID: 1, At: "files", Mode: workspace.ModeWrite}}} + _, _, err = a.CommitUpper(ctx, v, actor, 0, workspace.FixtureSource{workspace.FixtureFile("files/new", "content")}, workspace.UpperOptions{}) + } + done <- err + }() + <-ba.uploaded + writer := make(chan error, 1) + go func() { _, err := b.Mkdir(ctx, actor, "/blocked"); writer <- err }() + waitSharedLock(ba.pool) + Consistently(writer, 50*time.Millisecond).ShouldNot(Receive()) + close(ba.resume) + Expect(<-done).To(Succeed()) + Expect(<-writer).To(Succeed()) + }) + } + +}) + +// Observe PostgreSQL's actual lock wait rather than inferring it from sleeps. +func waitSharedLock(pool *sql.DB) { + GinkgoHelper() + Eventually(func() (bool, error) { + var blocked bool + err := pool.QueryRow(`SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE '%studio_ws_meta%')`).Scan(&blocked) + return blocked, err + }, 5*time.Second, 10*time.Millisecond).Should(BeTrue()) +} + +type cancelUpload struct { + ctx context.Context + entered chan struct{} +} + +func (r *cancelUpload) Read([]byte) (int, error) { + close(r.entered) + <-r.ctx.Done() + return 0, r.ctx.Err() +} diff --git a/core/services/studio/workspace/gormstore/store.go b/core/services/studio/workspace/gormstore/store.go index 19bd06fa9..840880dae 100644 --- a/core/services/studio/workspace/gormstore/store.go +++ b/core/services/studio/workspace/gormstore/store.go @@ -46,12 +46,13 @@ type Options struct { // Workspace is a workspace.Store. type Workspace struct { - db *conn - blobs blob.Store - admin string - maxSize int64 - now func() time.Time - grace time.Duration + db *conn + blobs blob.Store + admin string + maxSize int64 + now func() time.Time + grace time.Duration + sharedFilesystem bool } var ( @@ -87,7 +88,8 @@ func New(db *gorm.DB, blobs blob.Store, o Options) (*Workspace, error) { if now == nil { now = time.Now } - return &Workspace{db: c, blobs: blobs, admin: o.Admin, maxSize: o.MaxFileSize, now: now, grace: o.BlobGrace}, nil + fs, direct := blobs.(blob.Filesystem) + return &Workspace{db: c, blobs: blobs, admin: o.Admin, maxSize: o.MaxFileSize, now: now, grace: o.BlobGrace, sharedFilesystem: direct && fs.IsFilesystem() && d.name == "postgres"}, nil } // limit returns the smaller non-zero of the store limit and a per-call limit. diff --git a/core/services/studio/workspace/gormstore/upper.go b/core/services/studio/workspace/gormstore/upper.go index 4e2a0108f..44d7b3a49 100644 --- a/core/services/studio/workspace/gormstore/upper.go +++ b/core/services/studio/workspace/gormstore/upper.go @@ -67,15 +67,25 @@ func (s *Workspace) CommitUpper(ctx context.Context, v View, a Actor, base int64 if err != nil { return ChangeSet{}, nil, err } + var w *writer + if s.sharedFilesystem { + w, err = s.beginWrite(ctx, a, o.Title, ChangeAgent) + if err != nil { + return ChangeSet{}, nil, err + } + defer w.rollback() + } u, err := upperset.Read(ctx, v, pts, gone0, src, s.blobs, s.maxSize) if err != nil { return ChangeSet{}, nil, err } - w, err := s.beginWrite(ctx, a, o.Title, ChangeAgent) - if err != nil { - return ChangeSet{}, nil, err + if w == nil { + w, err = s.beginWrite(ctx, a, o.Title, ChangeAgent) + if err != nil { + return ChangeSet{}, nil, err + } + defer w.rollback() } - defer w.rollback() own, err := ownRevisions(ctx, w.tx, a, o.Own) if err != nil { return ChangeSet{}, nil, err diff --git a/docs/content/features/chats.md b/docs/content/features/chats.md index 4e0f9e8e7..99c214006 100644 --- a/docs/content/features/chats.md +++ b/docs/content/features/chats.md @@ -1646,11 +1646,12 @@ everything else in the chats works. A chat then has no skills either, not even the built-in ones, and no `load_skill`: skills are files, and every built-in skill works on files or runs commands. -In a cluster the bytes go to the object store, so a file written through one -frontend is read through another. That needs S3 (the storage URL setting); `docker-compose.distributed.yaml` sets it up with Garage. -Without it the object store is a folder of each frontend, so files are off in -that cluster, with a reason that says so, instead of every frontend keeping its -own copy. +In a cluster, configured S3 holds workspace bytes; the distributed compose +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. ## Storage @@ -1876,3 +1877,60 @@ The old chat page, the agents, the agent jobs and the old skills were replaced b The routes are listed under the `chats` tag at `/swagger/index.html`, and `GET /api/instructions/chats` returns a machine-readable summary. + +### Shared filesystem for Files + +Files selects its blob backend automatically, with no separate workspace path or +coordination flag: + +| Existing configuration | Workspace bytes | +| --- | --- | +| S3 storage URL configured | S3, using the existing bucket, region and credentials, for every role | +| No S3 storage URL | Direct filesystem storage at `DataPath/chats/blobs` | + +This applies to standalone servers, frontends, optional chat workers and the +add-on owner. An owner honors configured S3 without joining the distributed +control plane or changing its role. Global model/input staging is unchanged. +Standalone JSON/SQLite filesystem behavior and existing blob paths remain stable. +Without a data path and without S3, Files is disabled; temporary scratch never +becomes authoritative filesystem storage. With S3 and a database, local cache may +be temporary and is removed at shutdown. JSON metadata still needs a data path. +Configured S3 errors and PostgreSQL workspace initialization errors fail startup +rather than selecting another backend. + +Distributed Files requires shared PostgreSQL metadata. PostgreSQL owns names, +authorization, revisions and blob references; immutable content-addressed bytes +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. +Other data-path contents do not thereby become cluster-safe. + +Keep the canonical backing directory service-controlled, with compatible UID/GID +and permissions for every participant. LocalAI does not recursively chmod/chown +existing trees. Ordinary platform path aliases are resolved, while symlinks inside +the canonical blob tree are rejected; do not allow untrusted writers to replace +paths or redirect mounts. A same-node RWO/local-path volume needs enforced +co-placement and provides no cross-node failover guarantee. Multi-node use needs +coherent shared storage, normally an appropriate RWX volume: verify exclusive +creation, atomic same-filesystem rename, immediate cross-client visibility, file +sync and directory sync durability. An RWX label alone does not guarantee these +semantics; validate crash/power-loss behavior with the storage provider. + +All PostgreSQL-backed direct-filesystem workspaces automatically hold the revision +lock through blob ingest and metadata publication, including non-distributed +owners. This serializes metadata writes during slow uploads. Blocked operations +must honor request context cancellation. Orphan cleanup uses a bounded 30-second +context; cancellation can defer cleanup to a later pass. S3 is an alternative to +this shared-filesystem deployment contract; its local cache is not treated as a +shared filesystem. + +{{% notice warning %}} +No paths or blobs migrate automatically. Existing local data stays in place and +existing S3-backed workspaces stay on S3. Switching populated metadata to another +backend requires a separately planned blob copy and verification procedure, with +all participants switching consistently. Mixed S3/local/shared backends on the +same metadata can produce missing content. +{{% /notice %}} diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index 46857f08b..0e3a1629f 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -1584,3 +1584,39 @@ Install coalescing is per frontend instance and includes a keyed identity of the snapshot, so concurrent installs with different environments do not share an install reply. The worker owns process-local snapshots; they are rebuilt from model configuration on each install, not persisted as a new cluster state store. + +### Workspace Files without S3 + +Workspace Files uses configured S3 for every role; otherwise it uses the direct +filesystem adapter at the existing `DataPath/chats/blobs`. There is no additional +workspace path or boolean setting. This selection does not change global +model/input staging. The add-on owner honors the same S3 settings even without +distributed services, and remains an owner rather than joining the control plane. + +Distributed metadata still requires the same shared PostgreSQL database. Without +S3, mount the same existing DataPath/backing storage on every Files consumer: +frontends, the add-on owner and optional chat workers. Ordinary inference workers +need no Files mount. PostgreSQL owns metadata and references; immutable blobs are +shared, while scratch and caches are local and disposable. No DataPath and no S3 +means Files is disabled, not backed by authoritative temporary storage. + +Use compatible UID/GID, permissions and trusted writers. LocalAI does not +recursively chmod/chown existing trees. Platform path aliases are resolved, but +symlinks inside the canonical blob tree are rejected. Same-node local/RWO storage +requires enforced co-placement, with no cross-node failover claim. Multi-node use +requires coherent shared storage (typically RWX), atomic same-filesystem rename, +cross-client visibility and durable file/directory sync semantics. Equal path +strings do not prove shared mounts. Shared volume and datastore identity are not +verified at startup; no startup probe certifies these deployment guarantees. + +PostgreSQL/direct-FS publication and purge coordination is automatic for all +participants, including the separate owner. Blob ingest holds the revision lock, +serializing metadata writers; blocked operations must honor cancellation and +orphan cleanup has a bounded 30-second context. S3 remains an alternative. See +[Shared filesystem for Files](/features/chats/#shared-filesystem-for-files) for the +full storage and durability contract. + +No automatic migration occurs: existing filesystem paths stay put and configured +S3 workspaces remain on S3. Switching populated metadata requires a separately +planned copy and verification, with all consumers switching together. Mixed +S3/local/shared backends can yield missing content.