fix(distributed): expire staged request files on the worker

A request that carries a file stages it to the worker, which writes it
under its staging directory. Nothing removed it afterwards. The frontend
expires ephemeral keys from object storage, but that sweep never covered
a worker's local disk, so every image, audio clip and video a worker
ever served stayed on it.

One worker had accumulated 175 request directories over three months.
The volume reached 100 percent, and from that point every backend start
failed because the process manager could not create a state directory.

The worker now sweeps its ephemeral staging directory on a timer and
once at startup, so files left by a crash are reclaimed too. Staged
model files live beside that directory and are not touched.

Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
Assisted-by: Claude Code:claude-opus-5 [golangci-lint]

core/services/worker/ephemeral_cleanup.go

@@ -0,0 +1,112 @@
+package worker
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/mudler/xlog"
+)
+
+const (
+	// defaultEphemeralStagingTTL bounds how long a staged request input can
+	// outlive the request that needed it. Inference reads these files while the
+	// request runs, so the window has to cover a slow multimodal request; it
+	// does not have to cover anything longer.
+	defaultEphemeralStagingTTL = 6 * time.Hour
+	// defaultEphemeralStagingSweep is how often the worker sweeps.
+	defaultEphemeralStagingSweep = 30 * time.Minute
+)
+
+// StartEphemeralStagingCleanup sweeps the worker's own staging directory for
+// per-request input files left behind by finished requests.
+//
+// The frontend already expires ephemeral keys from object storage
+// (services/storage.StartEphemeralCleanup), but a worker receives these files
+// over the file-transfer server and writes them to its local disk, where
+// nothing expired them. They accumulated for as long as the worker lived and
+// eventually filled the volume, at which point every backend start failed
+// because the process manager could no longer create a state directory.
+func StartEphemeralStagingCleanup(ctx context.Context, stagingDir string, ttl, interval time.Duration) {
+	if stagingDir == "" {
+		return
+	}
+	if ttl <= 0 {
+		ttl = defaultEphemeralStagingTTL
+	}
+	if interval <= 0 {
+		interval = defaultEphemeralStagingSweep
+	}
+
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		// Sweep once at startup: a worker that crashed with staged files leaves
+		// them behind, and waiting a full interval to reclaim that space is the
+		// case that hurts on a volume that is already close to full.
+		CleanEphemeralStaging(stagingDir, ttl)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				CleanEphemeralStaging(stagingDir, ttl)
+			}
+		}
+	}()
+
+	xlog.Info("Ephemeral staging cleanup started", "dir", stagingDir, "ttl", ttl, "interval", interval)
+}
+
+// CleanEphemeralStaging removes staged per-request directories older than ttl.
+// It only ever descends into <stagingDir>/ephemeral, so staged model weights,
+// which live alongside it and are not scratch, are never considered.
+func CleanEphemeralStaging(stagingDir string, ttl time.Duration) {
+	root := filepath.Join(stagingDir, "ephemeral")
+	categories, err := os.ReadDir(root)
+	if err != nil {
+		// A worker that has never served a file-bearing request has no
+		// ephemeral directory at all. That is the normal case, not a fault.
+		if !os.IsNotExist(err) {
+			xlog.Warn("Ephemeral staging cleanup: cannot read staging root", "dir", root, "error", err)
+		}
+		return
+	}
+
+	cutoff := time.Now().Add(-ttl)
+	removed := 0
+	for _, category := range categories {
+		if !category.IsDir() {
+			continue
+		}
+		categoryDir := filepath.Join(root, category.Name())
+		entries, err := os.ReadDir(categoryDir)
+		if err != nil {
+			xlog.Warn("Ephemeral staging cleanup: cannot read category", "dir", categoryDir, "error", err)
+			continue
+		}
+		for _, entry := range entries {
+			path := filepath.Join(categoryDir, entry.Name())
+			info, err := entry.Info()
+			if err != nil {
+				xlog.Warn("Ephemeral staging cleanup: cannot stat entry", "path", path, "error", err)
+				continue
+			}
+			// A request rewrites nothing after staging, so the entry's own
+			// modification time is when its request was served.
+			if !info.ModTime().Before(cutoff) {
+				continue
+			}
+			if err := os.RemoveAll(path); err != nil {
+				xlog.Warn("Ephemeral staging cleanup: cannot remove", "path", path, "error", err)
+				continue
+			}
+			removed++
+		}
+	}
+
+	if removed > 0 {
+		xlog.Info("Ephemeral staging cleanup removed stale request files", "count", removed, "dir", root)
+	}
+}

core/services/worker/ephemeral_cleanup_test.go

@@ -0,0 +1,58 @@
+package worker
+
+import (
+	"os"
+	"path/filepath"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Worker ephemeral staging cleanup", func() {
+	var stagingDir string
+
+	// mkEphemeral creates one staged request directory holding a file, and
+	// backdates both so the sweeper sees it as `age` old.
+	mkEphemeral := func(id string, age time.Duration) string {
+		dir := filepath.Join(stagingDir, "ephemeral", "inputs", id)
+		Expect(os.MkdirAll(dir, 0o750)).To(Succeed())
+		Expect(os.WriteFile(filepath.Join(dir, "payload.bin"), []byte("x"), 0o600)).To(Succeed())
+		stamp := time.Now().Add(-age)
+		Expect(os.Chtimes(filepath.Join(dir, "payload.bin"), stamp, stamp)).To(Succeed())
+		Expect(os.Chtimes(dir, stamp, stamp)).To(Succeed())
+		return dir
+	}
+
+	BeforeEach(func() { stagingDir = GinkgoT().TempDir() })
+
+	It("removes staged request directories older than the TTL", func() {
+		old := mkEphemeral("aaaa1111", 48*time.Hour)
+		CleanEphemeralStaging(stagingDir, time.Hour)
+		Expect(old).ToNot(BeAnExistingFile())
+	})
+
+	It("keeps directories a running request may still be reading", func() {
+		fresh := mkEphemeral("bbbb2222", 5*time.Minute)
+		CleanEphemeralStaging(stagingDir, time.Hour)
+		Expect(fresh).To(BeAnExistingFile())
+	})
+
+	It("leaves staged models and everything outside ephemeral alone", func() {
+		modelDir := filepath.Join(stagingDir, "models", "some-model")
+		Expect(os.MkdirAll(modelDir, 0o750)).To(Succeed())
+		weights := filepath.Join(modelDir, "weights.gguf")
+		Expect(os.WriteFile(weights, []byte("w"), 0o600)).To(Succeed())
+		stamp := time.Now().Add(-90 * 24 * time.Hour)
+		Expect(os.Chtimes(weights, stamp, stamp)).To(Succeed())
+		Expect(os.Chtimes(modelDir, stamp, stamp)).To(Succeed())
+
+		CleanEphemeralStaging(stagingDir, time.Hour)
+
+		Expect(weights).To(BeAnExistingFile(), "a staged model is not ephemeral scratch")
+	})
+
+	It("does nothing when no ephemeral directory exists", func() {
+		Expect(func() { CleanEphemeralStaging(stagingDir, time.Hour) }).ToNot(Panic())
+	})
+})

core/services/worker/worker.go

@@ -159,6 +159,10 @@ func Run(ctx *cliContext.Context, cfg *Config) error {
 		return fmt.Errorf("starting HTTP file transfer server: %w", err)
 	}
 
+	// Per-request input files land in stagingDir over that server and nothing
+	// used to remove them, so a long-lived worker filled its own disk.
+	StartEphemeralStagingCleanup(shutdownCtx, stagingDir, 0, 0)
+
 	// Connect to NATS
 	xlog.Info("Connecting to NATS", "url", sanitize.URL(cfg.NatsURL))
 	natsClient, err := connectNats()

docs/content/features/distributed-mode.md

@@ -1020,6 +1020,12 @@ Notes:
 - Upgrade the worker when it does not support the exact model-stop request.
 - Stop and restart the stale backend only as an operational recovery action. LocalAI keeps it non-routable while durable cleanup is pending.
 
+**A worker fills its own disk over time:**
+- A request that carries a file (an image, an audio clip, a video) stages that file to the worker under `<models>/../staging/ephemeral/`. The worker deletes these 6 hours after the request that needed them, and sweeps every 30 minutes plus once at startup, so a worker that crashed mid-request still reclaims the space.
+- Releases before this sweep existed kept every staged input for the lifetime of the worker. Delete `<models>/../staging/ephemeral/` on an affected worker once, as the user the worker runs as; the sweep keeps it bounded from then on.
+- Staged **model** files are not touched by this. They live beside the ephemeral directory and are not per-request scratch.
+- A worker whose volume is genuinely full reports `creating backend process state directory under ...: no space left on device` when a backend starts.
+
 **Requests fail with `stale model config revision` although nobody edited the model:**
 - A model's stored revision must describe its persisted configuration. Releases before this fix also hashed the per-request prediction parameters, so the first request after a restart pinned the revision to its own `temperature`, `top_p`, `stop` and similar values. Every later request that sent different values was then rejected.
 - Upgrade the frontend replicas first. After the upgrade the revision is stamped when the configuration is loaded, so it no longer depends on the request body.