diff --git a/utils/cache/file_caches.go b/utils/cache/file_caches.go index 9788926d5..ed2374696 100644 --- a/utils/cache/file_caches.go +++ b/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 } diff --git a/utils/cache/file_caches_test.go b/utils/cache/file_caches_test.go index 9a9a9444f..edcfbc6b9 100644 --- a/utils/cache/file_caches_test.go +++ b/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 +} diff --git a/utils/cache/spread_fs.go b/utils/cache/spread_fs.go index 281e2bfab..647439790 100644 --- a/utils/cache/spread_fs.go +++ b/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) } diff --git a/utils/cache/spread_fs_test.go b/utils/cache/spread_fs_test.go index 2768ea2d5..0f88d3a58 100644 --- a/utils/cache/spread_fs_test.go +++ b/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()) }) }) })