fix(cache): don't serve partially-written transcodes after a crash (#5657)

* feat(cache): add completion marker helpers to spreadFS

* feat(cache): write completion marker after successful cache write

* fix(cache): adopt only complete files on reload, grandfather existing caches

* test(cache): regression tests for partial-transcode crash leftover (#5636)

* test(cache): guard concurrent in-progress streaming with completion marker

* test(cache): make concurrent-streaming guard actually attach a second reader mid-write

The previous test obtained s2 only after pw.Close(), so no reader ever
attached to the in-progress entry. Now pw.Write("hello ") is called
synchronously before the second Get — io.Pipe's blocking write gives a
deterministic happens-before — then both s1 and s2 are drained in parallel
goroutines while the producer writes the rest and closes the pipe.

* style(cache): clarify best-effort intent of cleanup os.Remove calls

* refactor(cache): lift one-time grandfather pass out of Reload's steady-state loop

* refactor(cache): have MarkComplete take the key, owning path mapping in spreadFS

* test(cache): assert no completion marker is written when the write fails

* refactor(cache): rename migration sentinel to generic .nd-migrated

* refactor(cache): rename grandfather migration to migrateExistingFiles

* refactor(cache): single-pass Reload with safer marker-error handling

Address PR review feedback:
- Merge the one-time migration into Reload's single directory walk,
  avoiding a second full walk on first boot.
- Only delete a data file when its marker is definitively absent
  (os.IsNotExist); skip on other stat errors to avoid destroying valid
  entries under transient I/O failures.
- Write the migration sentinel only after a clean walk, so a partial
  walk can't strand valid-but-unmarked files for later deletion.
- Return early from walkDataFiles on a WalkDir error.
- Assert fs.Create error in the marker-removal test.

utils/cache/file_caches.go

@@ -85,10 +85,11 @@ func NewFileCache(name, cacheSize, cacheFolder string, maxItems int, getReader R
 
 	go func() {
 		start := time.Now()
-		cache, err := newFSCache(fc.name, fc.cacheSize, fc.cacheFolder, fc.maxItems)
+		cache, sfs, err := newFSCache(fc.name, fc.cacheSize, fc.cacheFolder, fc.maxItems)
 		fc.mutex.Lock()
 		defer fc.mutex.Unlock()
 		fc.cache = cache
+		fc.fs = sfs
 		fc.disabled = cache == nil || err != nil
 		log.Info("Finished initializing cache", "cache", fc.name, "maxSize", fc.cacheSize, "elapsedTime", time.Since(start))
 		fc.ready.Store(true)
@@ -109,6 +110,7 @@ type fileCache struct {
 	cacheFolder string
 	maxItems    int
 	cache       fscache.Cache
+	fs          *spreadFS
 	getReader   ReadFunc
 	disabled    bool
 	ready       atomic.Bool
@@ -177,6 +179,7 @@ func (fc *fileCache) Get(ctx context.Context, arg Item) (*CachedStream, error) {
 				_ = fc.invalidate(ctx, key)
 			} else {
 				log.Trace(ctx, "File successfully stored in cache", "cache", fc.name, "key", key)
+				fc.markComplete(ctx, key)
 			}
 		}()
 	}
@@ -248,7 +251,18 @@ func copyAndClose(w io.WriteCloser, r io.Reader) error {
 	return err
 }
 
-func newFSCache(name, cacheSize, cacheFolder string, maxItems int) (fscache.Cache, error) {
+// markComplete records on disk that the entry for key was written in full,
+// so it is eligible for adoption after a restart (see spreadFS.Reload).
+func (fc *fileCache) markComplete(ctx context.Context, key string) {
+	if fc.fs == nil {
+		return
+	}
+	if err := fc.fs.MarkComplete(key); err != nil {
+		log.Warn(ctx, "Error writing cache completion marker", "cache", fc.name, "key", key, err)
+	}
+}
+
+func newFSCache(name, cacheSize, cacheFolder string, maxItems int) (fscache.Cache, *spreadFS, error) {
 	size, err := humanize.ParseBytes(cacheSize)
 	if err != nil {
 		log.Error("Invalid cache size. Using default size", "cache", name, "size", cacheSize,
@@ -257,7 +271,7 @@ func newFSCache(name, cacheSize, cacheFolder string, maxItems int) (fscache.Cach
 	}
 	if size == 0 {
 		log.Warn(fmt.Sprintf("%s cache disabled", name))
-		return nil, nil
+		return nil, nil, nil
 	}
 
 	lru := NewFileHaunter(name, maxItems, size, consts.DefaultCacheCleanUpInterval)
@@ -269,15 +283,15 @@ func newFSCache(name, cacheSize, cacheFolder string, maxItems int) (fscache.Cach
 	fs, err = NewSpreadFS(cacheFolder, 0755)
 	if err != nil {
 		log.Error(fmt.Sprintf("Error initializing %s cache FS", name), err)
-		return nil, err
+		return nil, nil, err
 	}
 
 	ck, err := fscache.NewCacheWithHaunter(fs, h)
 	if err != nil {
 		log.Error(fmt.Sprintf("Error initializing %s cache", name), err)
-		return nil, err
+		return nil, nil, err
 	}
 	ck.SetKeyMapper(fs.KeyMapper)
 
-	return ck, nil
+	return ck, fs, nil
 }

utils/cache/file_caches_test.go

@@ -104,6 +104,75 @@ var _ = Describe("File Caches", func() {
 			Expect(called).To(BeTrue())
 		})
 
+		It("writes a completion marker after a successful cache write", func() {
+			fc := callNewFileCache("test", "1KB", "test", 0, func(ctx context.Context, arg Item) (io.Reader, error) {
+				return strings.NewReader("complete-data"), nil
+			})
+			s, err := fc.Get(context.Background(), &testArg{"markme"})
+			Expect(err).To(BeNil())
+			_, _ = io.ReadAll(s)
+			_ = s.Close()
+
+			dataPath := fcSpreadFS(fc).KeyMapper((&testArg{"markme"}).Key())
+			Eventually(func() bool {
+				_, statErr := os.Stat(dataPath + ".complete")
+				return statErr == nil
+			}).Should(BeTrue())
+		})
+
+		It("serves a concurrent reader from an in-progress write and marks complete once", func() {
+			pr, pw := io.Pipe()
+			fc := callNewFileCache("test", "10MB", "test", 0, func(ctx context.Context, arg Item) (io.Reader, error) {
+				return pr, nil // slow, still-being-produced stream
+			})
+
+			// First Get → MISS; the cache starts copying pr into the entry in a goroutine.
+			s1, err := fc.Get(context.Background(), &testArg{"live"})
+			Expect(err).To(BeNil())
+
+			// Write the first chunk so the entry exists with in-flight bytes,
+			// but leave the pipe open so the second reader can attach mid-stream.
+			// io.Pipe writes block until the cache goroutine reads them, giving us
+			// a deterministic happens-before: the entry is live before we call Get again.
+			_, err = pw.Write([]byte("hello "))
+			Expect(err).To(BeNil())
+
+			// Second Get while the pipe is still open → attaches to the in-progress entry.
+			s2, err := fc.Get(context.Background(), &testArg{"live"})
+			Expect(err).To(BeNil())
+
+			// Drain both readers concurrently; they race against the producer below.
+			ch1 := make(chan []byte, 1)
+			ch2 := make(chan []byte, 1)
+			go func() { b, _ := io.ReadAll(s1); ch1 <- b }()
+			go func() { b, _ := io.ReadAll(s2); ch2 <- b }()
+
+			// Deliver the rest of the stream and close; both draining goroutines must see it.
+			_, err = pw.Write([]byte("world"))
+			Expect(err).To(BeNil())
+			Expect(pw.Close()).To(Succeed())
+
+			Expect(string(<-ch1)).To(Equal("hello world"))
+			Expect(string(<-ch2)).To(Equal("hello world"))
+			_ = s1.Close()
+			_ = s2.Close()
+
+			// Exactly one completion marker must appear.
+			dataPath := fcSpreadFS(fc).KeyMapper((&testArg{"live"}).Key())
+			Eventually(func() bool {
+				_, e := os.Stat(dataPath + ".complete")
+				return e == nil
+			}).Should(BeTrue())
+
+			// Steady-state HIT: full data, Cached flag set.
+			s3, err := fc.Get(context.Background(), &testArg{"live"})
+			Expect(err).To(BeNil())
+			got3, _ := io.ReadAll(s3)
+			_ = s3.Close()
+			Expect(s3.Cached).To(BeTrue())
+			Expect(string(got3)).To(Equal("hello world"))
+		})
+
 		Context("reader errors", func() {
 			When("creating a reader fails", func() {
 				It("does not cache", func() {
@@ -138,6 +207,77 @@ var _ = Describe("File Caches", func() {
 				})
 			})
 		})
+
+		Context("crash leftover (issue #5636)", func() {
+			It("does not serve a partial file left on disk as a complete HIT", func() {
+				// First init: empties + writes the migration sentinel.
+				fc1 := callNewFileCache("test", "10MB", "test", 0, func(ctx context.Context, arg Item) (io.Reader, error) {
+					return strings.NewReader("UNUSED"), nil
+				})
+				_ = fc1
+
+				// Plant a partial file (no marker), simulating a killed process.
+				sfs, err := NewSpreadFS(filepath.Join(conf.Server.CacheFolder.String(), "test"), 0755)
+				Expect(err).To(BeNil())
+				partialPath := sfs.KeyMapper((&testArg{"track"}).Key())
+				Expect(os.MkdirAll(filepath.Dir(partialPath), 0755)).To(Succeed())
+				Expect(os.WriteFile(partialPath, []byte("PARTIAL"), 0600)).To(Succeed())
+
+				// "Restart": a fresh cache over the same folder (sentinel present → strict).
+				getReaderCalled := false
+				fc2 := callNewFileCache("test", "10MB", "test", 0, func(ctx context.Context, arg Item) (io.Reader, error) {
+					getReaderCalled = true
+					return strings.NewReader("FULL-TRANSCODE"), nil
+				})
+
+				s, err := fc2.Get(context.Background(), &testArg{"track"})
+				Expect(err).To(BeNil())
+				data, _ := io.ReadAll(s)
+				_ = s.Close()
+
+				Expect(getReaderCalled).To(BeTrue()) // re-transcoded, not served stale
+				Expect(string(data)).To(Equal("FULL-TRANSCODE"))
+			})
+		})
+
+		Context("live error path still invalidates", func() {
+			It("leaves no data file and no marker after a mid-stream reader error", func() {
+				fc := callNewFileCache("test", "10MB", "test", 0, func(ctx context.Context, arg Item) (io.Reader, error) {
+					return errFakeReader{errors.New("boom")}, nil
+				})
+				s, err := fc.Get(context.Background(), &testArg{"err"})
+				Expect(err).To(BeNil())
+				_, _ = io.Copy(io.Discard, s)
+				_ = s.Close()
+
+				dataPath := fcSpreadFS(fc).KeyMapper((&testArg{"err"}).Key())
+				Eventually(func() bool {
+					_, e1 := os.Stat(dataPath)
+					_, e2 := os.Stat(dataPath + ".complete")
+					return os.IsNotExist(e1) && os.IsNotExist(e2)
+				}).Should(BeTrue())
+			})
+
+			It("does not write a completion marker when the write fails after partial bytes", func() {
+				// Mimics a transcode that produces real output and then dies:
+				// the bytes land on disk, but the entry must NOT be marked complete.
+				fc := callNewFileCache("test", "10MB", "test", 0, func(ctx context.Context, arg Item) (io.Reader, error) {
+					return &partialThenErrReader{data: []byte("PARTIAL-OUTPUT"), err: errors.New("transcoder died")}, nil
+				})
+				s, err := fc.Get(context.Background(), &testArg{"partial"})
+				Expect(err).To(BeNil())
+				_, _ = io.Copy(io.Discard, s)
+				_ = s.Close()
+
+				dataPath := fcSpreadFS(fc).KeyMapper((&testArg{"partial"}).Key())
+				// The marker must never appear for a failed write. Give the async
+				// writer time to finish, then assert the marker stays absent.
+				Consistently(func() bool {
+					_, e := os.Stat(dataPath + ".complete")
+					return os.IsNotExist(e)
+				}).Should(BeTrue())
+			})
+		})
 	})
 })
 
@@ -148,3 +288,23 @@ func (t *testArg) Key() string { return t.s }
 type errFakeReader struct{ err error }
 
 func (e errFakeReader) Read([]byte) (int, error) { return 0, e.err }
+
+// partialThenErrReader emits data once, then fails — mimicking a transcoder
+// that produces some output and then dies mid-stream.
+type partialThenErrReader struct {
+	data []byte
+	err  error
+	done bool
+}
+
+func (r *partialThenErrReader) Read(p []byte) (int, error) {
+	if r.done {
+		return 0, r.err
+	}
+	r.done = true
+	return copy(p, r.data), nil
+}
+
+func fcSpreadFS(fc *fileCache) *spreadFS {
+	return fc.fs
+}

utils/cache/spread_fs.go

@@ -14,6 +14,9 @@ import (
 	"github.com/navidrome/navidrome/log"
 )
 
+const completeMarkerSuffix = ".complete"
+const sentinelName = ".nd-migrated"
+
 type spreadFS struct {
 	root string
 	mode os.FileMode
@@ -40,30 +43,83 @@ func NewSpreadFS(dir string, mode os.FileMode) (*spreadFS, error) {
 }
 
 func (sfs *spreadFS) Reload(f func(key string, name string)) error {
+	// On the first run after upgrade (no sentinel yet), pre-existing files have
+	// no completion marker. Migrate them instead of discarding them as partials,
+	// so a user's whole cache isn't wiped. After the sentinel exists, an unmarked
+	// file is a crash partial and is discarded.
+	sentinel := filepath.Join(sfs.root, sentinelName)
+	_, sErr := os.Stat(sentinel)
+	migrating := os.IsNotExist(sErr)
+
 	count := 0
-	err := filepath.WalkDir(sfs.root, func(absoluteFilePath string, de fs.DirEntry, err error) error {
+	err := sfs.walkDataFiles(func(absoluteFilePath string) {
+		if _, statErr := os.Stat(sfs.markerPath(absoluteFilePath)); statErr != nil {
+			switch {
+			case migrating:
+				if mErr := sfs.MarkComplete(absoluteFilePath); mErr != nil {
+					log.Warn("Error migrating cache file", "file", absoluteFilePath, mErr)
+				}
+			case os.IsNotExist(statErr):
+				// No completion marker: this is a partial left by a crash. Discard it.
+				log.Debug("Removing incomplete cache file", "file", absoluteFilePath)
+				_ = os.Remove(absoluteFilePath) //nolint:gosec // best-effort cleanup; re-swept on next Reload
+				return
+			default:
+				// Marker may exist but is unreadable (transient I/O, permissions):
+				// skip adoption without destroying a possibly-valid entry.
+				log.Warn("Error reading cache completion marker", "file", absoluteFilePath, statErr)
+				return
+			}
+		}
+		f(absoluteFilePath, absoluteFilePath)
+		count++
+	})
+	if err != nil {
+		return err
+	}
+
+	log.Debug("Loaded cache", "dir", sfs.root, "numItems", count)
+	// Only record the migration as done after a clean walk, so a partial walk
+	// doesn't leave valid-but-unmarked files to be discarded on the next run.
+	if migrating {
+		if wErr := os.WriteFile(sentinel, nil, 0600); wErr != nil {
+			log.Warn("Error writing cache migration sentinel", "file", sentinel, wErr)
+		}
+	}
+	return nil
+}
+
+// walkDataFiles visits every cache data file (named XX/XX/<40-hex>), skipping
+// completion markers and opportunistically cleaning up orphaned ones.
+func (sfs *spreadFS) walkDataFiles(visit func(absoluteFilePath string)) error {
+	return filepath.WalkDir(sfs.root, func(absoluteFilePath string, _ fs.DirEntry, err error) error {
 		if err != nil {
 			log.Error("Error loading cache", "dir", sfs.root, err)
+			return nil
 		}
 		path, err := filepath.Rel(sfs.root, absoluteFilePath)
 		if err != nil {
 			return nil //nolint:nilerr
 		}
 
+		// Skip marker files; also clean orphan markers (data file gone).
+		if strings.HasSuffix(path, completeMarkerSuffix) {
+			dataPath := strings.TrimSuffix(absoluteFilePath, completeMarkerSuffix)
+			if _, statErr := os.Stat(dataPath); os.IsNotExist(statErr) {
+				_ = os.Remove(absoluteFilePath) //nolint:gosec // best-effort cleanup; re-swept on next Reload
+			}
+			return nil
+		}
+
 		// Skip if name is not in the format XX/XX/XXXXXXXXXXXX
 		parts := strings.Split(path, string(os.PathSeparator))
 		if len(parts) != 3 || len(parts[0]) != 2 || len(parts[1]) != 2 || len(parts[2]) != 40 {
 			return nil
 		}
 
-		f(absoluteFilePath, absoluteFilePath)
-		count++
+		visit(absoluteFilePath)
 		return nil
 	})
-	if err == nil {
-		log.Debug("Loaded cache", "dir", sfs.root, "numItems", count)
-	}
-	return err
 }
 
 func (sfs *spreadFS) Create(name string) (stream.File, error) {
@@ -79,7 +135,27 @@ func (sfs *spreadFS) Open(name string) (stream.File, error) {
 	return os.Open(name)
 }
 
+func (sfs *spreadFS) markerPath(dataPath string) string {
+	return dataPath + completeMarkerSuffix
+}
+
+// MarkComplete records that the cache entry for key was written in full.
+// Only files with a marker are adopted on the next Reload; this is what
+// distinguishes a complete cache entry from a partial one left by a crash.
+// key may be an original cache key or an already-mapped data path; KeyMapper
+// is idempotent for the latter (see KeyMapper).
+func (sfs *spreadFS) MarkComplete(key string) error {
+	f, err := os.OpenFile(sfs.markerPath(sfs.KeyMapper(key)), os.O_CREATE|os.O_WRONLY, 0600)
+	if err != nil {
+		return err
+	}
+	return f.Close()
+}
+
 func (sfs *spreadFS) Remove(name string) error {
+	if err := os.Remove(sfs.markerPath(name)); err != nil && !os.IsNotExist(err) {
+		log.Warn("Error removing cache completion marker", "file", name, err)
+	}
 	return os.Remove(name)
 }
 

utils/cache/spread_fs_test.go

@@ -39,31 +39,93 @@ var _ = Describe("Spread FS", func() {
 		})
 	})
 
-	Describe("Reload", func() {
-		var files []string
+	Describe("MarkComplete / Remove markers", func() {
+		It("creates a .complete marker for a data file", func() {
+			data := fs.KeyMapper("song1")
+			f, err := fs.Create(data)
+			Expect(err).To(BeNil())
+			_, _ = f.Write([]byte("ok"))
+			_ = f.Close()
 
-		BeforeEach(func() {
-			files = []string{"aaaaa", "bbbbb", "ccccc"}
-			for _, content := range files {
-				file := fs.KeyMapper(content)
-				f, err := fs.Create(file)
-				Expect(err).To(BeNil())
-				_, _ = f.Write([]byte(content))
-				_ = f.Close()
-			}
+			Expect(fs.MarkComplete(data)).To(Succeed())
+			_, statErr := os.Stat(data + ".complete")
+			Expect(statErr).To(BeNil())
 		})
 
-		It("loads all files from fs", func() {
+		It("removes the sibling marker when the data file is removed", func() {
+			data := fs.KeyMapper("song2")
+			f, err := fs.Create(data)
+			Expect(err).To(BeNil())
+			_, _ = f.Write([]byte("ok"))
+			_ = f.Close()
+			Expect(fs.MarkComplete(data)).To(Succeed())
+
+			Expect(fs.Remove(data)).To(Succeed())
+			_, dataErr := os.Stat(data)
+			Expect(os.IsNotExist(dataErr)).To(BeTrue())
+			_, markErr := os.Stat(data + ".complete")
+			Expect(os.IsNotExist(markErr)).To(BeTrue())
+		})
+	})
+
+	Describe("Reload", func() {
+		makeData := func(content string) string {
+			file := fs.KeyMapper(content)
+			f, err := fs.Create(file)
+			Expect(err).To(BeNil())
+			_, _ = f.Write([]byte(content))
+			_ = f.Close()
+			return file
+		}
+
+		It("migrates all existing files on first run and writes the sentinel", func() {
+			for _, c := range []string{"aaaaa", "bbbbb", "ccccc"} {
+				makeData(c) // no markers, simulating a pre-upgrade cache
+			}
+
 			var actual []string
-			err := fs.Reload(func(key string, name string) {
+			err := fs.Reload(func(key, name string) {
 				Expect(key).To(Equal(name))
-				data, err := os.ReadFile(name)
-				Expect(err).To(BeNil())
+				data, _ := os.ReadFile(name)
 				actual = append(actual, string(data))
 			})
 			Expect(err).To(BeNil())
-			Expect(actual).To(HaveLen(len(files)))
-			Expect(actual).To(ContainElements(files[0], files[1], files[2]))
+			Expect(actual).To(ContainElements("aaaaa", "bbbbb", "ccccc"))
+			Expect(actual).To(HaveLen(3))
+
+			_, sentinelErr := os.Stat(filepath.Join(rootDir, ".nd-migrated"))
+			Expect(sentinelErr).To(BeNil())
+		})
+
+		It("after migration, adopts only marked files and deletes unmarked partials", func() {
+			// Pretend migration already happened.
+			Expect(os.WriteFile(filepath.Join(rootDir, ".nd-migrated"), nil, 0600)).To(Succeed())
+
+			good := makeData("good")
+			Expect(fs.MarkComplete(good)).To(Succeed())
+			bad := makeData("bad") // partial: no marker
+
+			var actual []string
+			err := fs.Reload(func(key, name string) { actual = append(actual, name) })
+			Expect(err).To(BeNil())
+			Expect(actual).To(ConsistOf(good))
+
+			_, badErr := os.Stat(bad)
+			Expect(os.IsNotExist(badErr)).To(BeTrue()) // partial deleted
+		})
+
+		It("ignores and cleans orphan markers", func() {
+			Expect(os.WriteFile(filepath.Join(rootDir, ".nd-migrated"), nil, 0600)).To(Succeed())
+			orphan := fs.KeyMapper("orphan") + ".complete"
+			Expect(os.MkdirAll(filepath.Dir(orphan), 0755)).To(Succeed())
+			Expect(os.WriteFile(orphan, nil, 0600)).To(Succeed())
+
+			var actual []string
+			err := fs.Reload(func(key, name string) { actual = append(actual, name) })
+			Expect(err).To(BeNil())
+			Expect(actual).To(BeEmpty())
+			_, orphanErr := os.Stat(orphan)
+			Expect(os.IsNotExist(orphanErr)).To(BeTrue())
 		})
 	})
 })