diff --git a/cmd/silo/main.go b/cmd/silo/main.go index 8c4da0a6..06e2a3ab 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -1967,6 +1967,9 @@ func main() { if deps.FolderRepo != nil && deps.LibraryScanQueue != nil { taskMgr.Register(tasks.NewScanLibrariesTask(deps.FolderRepo, deps.LibraryScanQueue, deps.EventBus)) } + if deps.FolderRepo != nil && deps.Scanner != nil { + taskMgr.Register(tasks.NewVerifyFilePresenceTask(deps.FolderRepo, deps.Scanner, 0)) + } taskMgr.Register(tasks.NewCleanupOrphanedMediaItemsTask(catalog.NewOrphanedProvisionalCleaner(deps.DB))) taskMgr.Register(tasks.NewBackfillMediaItemAliasesTask(catalog.NewItemAliasRepository(deps.DB))) if deps.S3Public != nil { @@ -2207,6 +2210,9 @@ func main() { if deps.ImageResolver != nil { absDetailSvc.SetImageResolver(deps.ImageResolver) } + if deps.FileRepo != nil { + absDetailSvc.SetMissingFileMarker(deps.FileRepo) + } var absScopeResolver scopeResolver if policySystem != nil { absScopeResolver = policy.NewViewerResolver(absUserRepo, userStoreProvider, nil, policySystem.PDP(), accessGroupStore) @@ -2499,6 +2505,9 @@ func main() { detailSvc.SetGroupClaimRepository(catalog.NewGroupClaimRepository(deps.DB)) detailSvc.SetProbeEnsurer(deps.ProbeEnsurer) detailSvc.SetChapterThumbnailQueuer(deps.ChapterThumbnailQueuer) + if deps.FileRepo != nil { + detailSvc.SetMissingFileMarker(deps.FileRepo) + } if deps.ImageResolver != nil { detailSvc.SetImageResolver(deps.ImageResolver) } diff --git a/internal/catalog/detail.go b/internal/catalog/detail.go index ab97aef9..2cbfad84 100644 --- a/internal/catalog/detail.go +++ b/internal/catalog/detail.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log/slog" + "os" "path/filepath" "sort" "strconv" @@ -614,6 +615,13 @@ type DetailService struct { originalLangFn func(context.Context, string) string probeEnsurer PlaybackProbeEnsurer chapterThumbs ChapterThumbnailQueuer + missingMarker MissingFileMarker +} + +// MissingFileMarker records that a file the catalog believed was present is +// not on disk. Mirrors the playback preflight's marker. +type MissingFileMarker interface { + MarkMissing(ctx context.Context, id int, since time.Time) error } // NewDetailService creates a new DetailService. @@ -663,6 +671,13 @@ func (s *DetailService) SetChapterThumbnailQueuer(queuer ChapterThumbnailQueuer) s.chapterThumbs = queuer } +// SetMissingFileMarker enables the presence check in preparePlaybackFiles. When +// unset the check still filters the response, it just cannot persist what it +// found, so the next request repeats the work. +func (s *DetailService) SetMissingFileMarker(marker MissingFileMarker) { + s.missingMarker = marker +} + func (s *DetailService) SetFolderRepository(repo interface { GetByID(ctx context.Context, id int) (*models.MediaFolder, error) }) { @@ -3320,6 +3335,8 @@ func (s *DetailService) preparePlaybackFiles(ctx context.Context, files []*model return files } + files = s.dropVanishedFiles(ctx, files) + prepared := make([]*models.MediaFile, 0, len(files)) for _, file := range files { if file == nil { @@ -3337,6 +3354,69 @@ func (s *DetailService) preparePlaybackFiles(ctx context.Context, files []*model return prepared } +// vanishedCheckBudget caps how long the presence check may spend on one +// request. Detail loads are on the interactive path, and a library root that +// has become slow rather than absent must not turn every detail view into a +// stall — past the budget the remaining files are returned unchecked and the +// background sweep picks them up. +const vanishedCheckBudget = 250 * time.Millisecond + +// vanishedCheckLimit caps how many files are checked per request. A movie has +// a handful of versions, but an audiobook item can carry hundreds of files, +// and the marginal value of checking the 300th one inline is nil. +const vanishedCheckLimit = 32 + +// dropVanishedFiles removes files that are no longer on disk and records the +// removal so every other read path sees it too. +// +// The scanner and the background presence sweep are what keep the catalog +// honest in bulk; this is the last check before a file is offered to a user as +// something they can play. Without it a file deleted since the last sweep is +// still listed as a version, and the user finds out by pressing play and +// getting an error — which is the failure this is here to prevent. +// +// Only os.ErrNotExist drops a file. Any other stat error means the file's +// state is unknown, and an unreadable-but-present file is still worth +// offering: the playback preflight will produce a real error if it is not. +func (s *DetailService) dropVanishedFiles(ctx context.Context, files []*models.MediaFile) []*models.MediaFile { + deadline := time.Now().Add(vanishedCheckBudget) + kept := make([]*models.MediaFile, 0, len(files)) + var vanished []*models.MediaFile + + for i, file := range files { + if file == nil { + continue + } + if i >= vanishedCheckLimit || time.Now().After(deadline) { + kept = append(kept, file) + continue + } + if _, err := os.Stat(file.FilePath); err != nil && errors.Is(err, os.ErrNotExist) { + vanished = append(vanished, file) + continue + } + kept = append(kept, file) + } + + if len(vanished) == 0 { + return files + } + + if s.missingMarker != nil { + since := time.Now().UTC() + for _, file := range vanished { + if err := s.missingMarker.MarkMissing(ctx, file.ID, since); err != nil { + slog.WarnContext(ctx, "failed to mark vanished file missing", "component", "catalog", + "file_id", file.ID, "path", file.FilePath, "error", err) + } + } + } + + slog.InfoContext(ctx, "catalog: hid files missing from disk", "component", "catalog", + "count", len(vanished), "remaining", len(kept)) + return kept +} + func (s *DetailService) queueWatchPlaybackFiles( ctx context.Context, contentID string, diff --git a/internal/catalog/detail_vanished_test.go b/internal/catalog/detail_vanished_test.go new file mode 100644 index 00000000..7528145d --- /dev/null +++ b/internal/catalog/detail_vanished_test.go @@ -0,0 +1,162 @@ +package catalog + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/models" +) + +type recordingMissingMarker struct { + marked []int + err error +} + +func (m *recordingMissingMarker) MarkMissing(_ context.Context, id int, _ time.Time) error { + m.marked = append(m.marked, id) + return m.err +} + +func fileAt(id int, path string) *models.MediaFile { + return &models.MediaFile{ID: id, FilePath: path} +} + +func TestDropVanishedFilesHidesRemovedVersions(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + live := filepath.Join(dir, "movie.2160p.mkv") + if err := os.WriteFile(live, []byte("x"), 0o644); err != nil { + t.Fatalf("writing live file: %v", err) + } + + marker := &recordingMissingMarker{} + svc := &DetailService{missingMarker: marker} + + // The upgrade case: an older release was replaced on disk but its row is + // still live because no scan has run since. + got := svc.dropVanishedFiles(context.Background(), []*models.MediaFile{ + fileAt(1, filepath.Join(dir, "movie.1080p.mkv")), + fileAt(2, live), + }) + + if len(got) != 1 || got[0].ID != 2 { + t.Fatalf("kept %d files (first id %v), want only the file that exists on disk", len(got), fileIDsOf(got)) + } + if len(marker.marked) != 1 || marker.marked[0] != 1 { + t.Fatalf("marked = %v, want [1]", marker.marked) + } +} + +func TestDropVanishedFilesKeepsEverythingWhenNothingVanished(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + paths := []string{filepath.Join(dir, "a.mkv"), filepath.Join(dir, "b.mkv")} + for _, p := range paths { + if err := os.WriteFile(p, []byte("x"), 0o644); err != nil { + t.Fatalf("writing %s: %v", p, err) + } + } + + marker := &recordingMissingMarker{} + svc := &DetailService{missingMarker: marker} + files := []*models.MediaFile{fileAt(1, paths[0]), fileAt(2, paths[1])} + + got := svc.dropVanishedFiles(context.Background(), files) + if len(got) != 2 { + t.Fatalf("kept %d files, want 2", len(got)) + } + if len(marker.marked) != 0 { + t.Fatalf("marked = %v, want none", marker.marked) + } +} + +// A failing MarkMissing must not resurrect the file in the response: the user +// still cannot play it, so listing it would reintroduce the exact dead-click +// this check exists to prevent. +func TestDropVanishedFilesHidesEvenWhenMarkingFails(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + marker := &recordingMissingMarker{err: errors.New("database down")} + svc := &DetailService{missingMarker: marker} + + got := svc.dropVanishedFiles(context.Background(), []*models.MediaFile{ + fileAt(1, filepath.Join(dir, "gone.mkv")), + }) + if len(got) != 0 { + t.Fatalf("kept %d files, want 0", len(got)) + } +} + +// The marker is optional wiring; the filter must still work without it. +func TestDropVanishedFilesWorksWithoutMarker(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + svc := &DetailService{} + got := svc.dropVanishedFiles(context.Background(), []*models.MediaFile{ + fileAt(1, filepath.Join(dir, "gone.mkv")), + }) + if len(got) != 0 { + t.Fatalf("kept %d files, want 0", len(got)) + } +} + +// Past the per-request limit files pass through unchecked rather than making +// an item with hundreds of files pay for hundreds of stats on every load. +func TestDropVanishedFilesRespectsCheckLimit(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + files := make([]*models.MediaFile, 0, vanishedCheckLimit+5) + for i := range vanishedCheckLimit + 5 { + files = append(files, fileAt(i+1, filepath.Join(dir, "gone.mkv"))) + } + + marker := &recordingMissingMarker{} + svc := &DetailService{missingMarker: marker} + got := svc.dropVanishedFiles(context.Background(), files) + + if len(got) != 5 { + t.Fatalf("kept %d files, want 5 (the ones past the check limit)", len(got)) + } + if len(marker.marked) != vanishedCheckLimit { + t.Fatalf("marked %d files, want %d", len(marker.marked), vanishedCheckLimit) + } +} + +func TestPreparePlaybackFilesDropsVanished(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + live := filepath.Join(dir, "live.mkv") + if err := os.WriteFile(live, []byte("x"), 0o644); err != nil { + t.Fatalf("writing live file: %v", err) + } + + svc := &DetailService{missingMarker: &recordingMissingMarker{}} + got := svc.preparePlaybackFiles(context.Background(), []*models.MediaFile{ + fileAt(1, filepath.Join(dir, "gone.mkv")), + fileAt(2, live), + }) + + if len(got) != 1 || got[0].ID != 2 { + t.Fatalf("preparePlaybackFiles returned %v, want only file 2", fileIDsOf(got)) + } +} + +func fileIDsOf(files []*models.MediaFile) []int { + ids := make([]int, 0, len(files)) + for _, f := range files { + if f != nil { + ids = append(ids, f.ID) + } + } + return ids +} diff --git a/internal/scanner/file_repo.go b/internal/scanner/file_repo.go index cc46f0ab..434da4c8 100644 --- a/internal/scanner/file_repo.go +++ b/internal/scanner/file_repo.go @@ -2640,6 +2640,59 @@ func claimRepresentativeWindow(limit int) int { return window } +// PresentFileRef is the minimal projection the presence sweep needs: enough to +// stat a file and mark it missing. Deliberately not the full media_files row — +// the sweep walks every live file in a library, so the per-row cost matters. +type PresentFileRef struct { + ID int + Path string +} + +// ListPresentByFolder returns id/path for every file in the folder that the +// catalog currently believes is on disk (missing_since IS NULL). Ordered by id +// so a sweep interrupted partway through covers a stable prefix. +func (r *FileRepository) ListPresentByFolder(ctx context.Context, folderID int) ([]PresentFileRef, error) { + rows, err := r.pool.Query(ctx, + `SELECT id, file_path FROM media_files + WHERE media_folder_id = $1 AND missing_since IS NULL + ORDER BY id ASC`, folderID) + if err != nil { + return nil, fmt.Errorf("listing present files for folder %d: %w", folderID, err) + } + defer rows.Close() + + refs := make([]PresentFileRef, 0) + for rows.Next() { + var ref PresentFileRef + if err := rows.Scan(&ref.ID, &ref.Path); err != nil { + return nil, fmt.Errorf("scanning present file for folder %d: %w", folderID, err) + } + refs = append(refs, ref) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating present files for folder %d: %w", folderID, err) + } + return refs, nil +} + +// MarkMissingByIDs sets missing_since on every given file that is not already +// marked, in one statement. Returns the number of rows newly marked; ids that +// were already missing (a concurrent scan beat us to them) are not counted and +// keep their original timestamp, so the removal grace is measured from first +// detection rather than being extended by every later sweep. +func (r *FileRepository) MarkMissingByIDs(ctx context.Context, ids []int, since time.Time) (int, error) { + if len(ids) == 0 { + return 0, nil + } + tag, err := r.pool.Exec(ctx, + `UPDATE media_files SET missing_since = $1, updated_at = NOW() + WHERE id = ANY($2) AND missing_since IS NULL`, since, ids) + if err != nil { + return 0, fmt.Errorf("marking files missing: %w", err) + } + return int(tag.RowsAffected()), nil +} + // MarkMissing sets the missing_since timestamp for the given media file. func (r *FileRepository) MarkMissing(ctx context.Context, id int, since time.Time) error { tag, err := r.pool.Exec(ctx, diff --git a/internal/scanner/presence_sweep.go b/internal/scanner/presence_sweep.go new file mode 100644 index 00000000..0659644b --- /dev/null +++ b/internal/scanner/presence_sweep.go @@ -0,0 +1,211 @@ +package scanner + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "sync" + "time" + + "github.com/Silo-Server/silo-server/internal/models" +) + +// presenceSweepWorkers bounds the concurrent stat calls. A sweep is pure +// syscall work with no ffprobe or hashing, but library roots are frequently +// network mounts where a wide fan-out buys latency at the cost of hammering +// the server, so keep the fan-out modest. +const presenceSweepWorkers = 16 + +// presenceSweepAbortFraction is the share of a library's live files that may +// vanish in a single sweep before the sweep refuses to act on the result. +// +// The sweep runs unattended every few minutes and only ever observes the +// filesystem through stat(2), so it cannot distinguish "the operator deleted +// half the library" from "a mount dropped out but left a populated-looking +// directory behind". The root probes upstream catch the common shapes of that +// fault; this is the backstop for the ones they miss. Bulk removals are real, +// but they are also the case where being wrong is most expensive, so the sweep +// defers them to a full scan rather than guessing. +const presenceSweepAbortFraction = 0.5 + +// presenceSweepAbortFloor is the number of vanished files below which the +// fraction check does not apply. Without it a library with two files would +// abort on a single legitimate deletion. +const presenceSweepAbortFloor = 20 + +// PresenceSweepResult reports what one folder's sweep observed. +type PresenceSweepResult struct { + Checked int + Vanished int + Marked int + // Skipped counts live files under a protected (unreachable or + // suspect-empty) root, which the sweep does not stat at all. + Skipped int + // Aborted is set when the vanished count tripped the bulk-removal + // backstop and nothing was marked for this folder. + Aborted bool + // MembershipsRemoved and ItemsDeleted come from the reconcile pass that + // follows a sweep that marked anything. + MembershipsRemoved int + ItemsDeleted int +} + +// VerifyPresence stats every file the catalog believes is on disk for one +// folder and marks the vanished ones missing, then reconciles library +// membership so titles left with no playable file drop out of the catalog. +// +// This exists because a deleted file is otherwise invisible to the server +// until the next full library scan: catalog reads all filter on +// missing_since, so the row keeps advertising a playable version, and the +// first thing to notice is a user pressing play. A sweep is cheap enough +// (one stat per live file, no probing) to run on a short interval, which +// closes that window from a scan interval down to the sweep interval. +// +// It deliberately does not empty trash. Deciding that a row is gone for good +// is the full scan's job, which has walked the tree and knows what is +// actually there; the sweep only knows that specific paths did not resolve. +func (s *Scanner) VerifyPresence(ctx context.Context, folder *models.MediaFolder) (PresenceSweepResult, error) { + var result PresenceSweepResult + if s == nil || s.fileRepo == nil || folder == nil { + return result, fmt.Errorf("verify presence: scanner not configured") + } + + // Protect files under roots that are unreachable or suspect-empty: those + // files have not been removed, their storage is just not answering, and + // stat would report every one of them as gone. + configuredPaths, err := s.configuredFolderPaths(ctx, folder) + if err != nil { + return result, err + } + configuredRoots := cleanScanRoots(configuredPaths) + protectedRoots := probeUnreachableRoots(ctx, folder.ID, configuredRoots) + suspectRoots, err := s.suspectEmptyRoots(ctx, folder.ID, configuredRoots, protectedRoots) + if err != nil { + return result, err + } + protectedRoots = append(protectedRoots, suspectRoots...) + + refs, err := s.fileRepo.ListPresentByFolder(ctx, folder.ID) + if err != nil { + return result, err + } + if len(refs) == 0 { + return result, nil + } + + candidates := make([]PresentFileRef, 0, len(refs)) + for _, ref := range refs { + if len(protectedRoots) > 0 && pathWithinAnyRoot(ref.Path, protectedRoots) { + result.Skipped++ + continue + } + candidates = append(candidates, ref) + } + result.Checked = len(candidates) + if len(candidates) == 0 { + return result, nil + } + + vanished, err := statVanished(ctx, candidates) + if err != nil { + return result, err + } + result.Vanished = len(vanished) + if len(vanished) == 0 { + return result, nil + } + + if len(vanished) >= presenceSweepAbortFloor && + float64(len(vanished)) >= float64(result.Checked)*presenceSweepAbortFraction { + result.Aborted = true + slog.WarnContext(ctx, "scanner: presence sweep saw a bulk disappearance; deferring to a full scan", + "component", "scanner", + "folder_id", folder.ID, + "checked", result.Checked, + "vanished", result.Vanished, + ) + return result, nil + } + + marked, err := s.fileRepo.MarkMissingByIDs(ctx, vanished, time.Now().UTC()) + if err != nil { + return result, err + } + result.Marked = marked + if marked == 0 { + return result, nil + } + + removedMemberships, deletedItems, orphanedImageDirs, err := s.reconcileLibraryMemberships(ctx, folder.ID, protectedRoots) + if err != nil { + return result, fmt.Errorf("reconciling library membership for folder %d: %w", folder.ID, err) + } + result.MembershipsRemoved = removedMemberships + result.ItemsDeleted = deletedItems + + if s.s3Client != nil && len(orphanedImageDirs) > 0 { + bucket := s.s3Client.Bucket() + for _, dir := range orphanedImageDirs { + _, _ = s.s3Client.DeletePrefix(ctx, bucket, dir) + } + } + + slog.InfoContext(ctx, "scanner: presence sweep marked files missing", "component", "scanner", + "folder_id", folder.ID, + "checked", result.Checked, + "marked", result.Marked, + "memberships_removed", result.MembershipsRemoved, + "items_deleted", result.ItemsDeleted, + ) + return result, nil +} + +// statVanished returns the ids of refs whose path no longer exists. Only +// os.ErrNotExist counts: a permission error, a timing-out mount, or any other +// fault means the file's state is unknown, and treating unknown as gone is +// what the sweep must never do. +func statVanished(ctx context.Context, refs []PresentFileRef) ([]int, error) { + workers := presenceSweepWorkers + if len(refs) < workers { + workers = len(refs) + } + + var ( + mu sync.Mutex + vanished []int + ) + jobs := make(chan PresentFileRef) + var wg sync.WaitGroup + + for range workers { + wg.Add(1) + go func() { + defer wg.Done() + for ref := range jobs { + if _, err := os.Stat(ref.Path); err == nil || !errors.Is(err, os.ErrNotExist) { + continue + } + mu.Lock() + vanished = append(vanished, ref.ID) + mu.Unlock() + } + }() + } + + var walkErr error + for _, ref := range refs { + if err := ctx.Err(); err != nil { + walkErr = err + break + } + jobs <- ref + } + close(jobs) + wg.Wait() + if walkErr != nil { + return nil, walkErr + } + return vanished, nil +} diff --git a/internal/scanner/presence_sweep_test.go b/internal/scanner/presence_sweep_test.go new file mode 100644 index 00000000..528cab79 --- /dev/null +++ b/internal/scanner/presence_sweep_test.go @@ -0,0 +1,136 @@ +package scanner + +import ( + "context" + "os" + "path/filepath" + "slices" + "testing" +) + +func TestStatVanishedReportsOnlyAbsentFiles(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + present := filepath.Join(dir, "present.mkv") + if err := os.WriteFile(present, []byte("x"), 0o644); err != nil { + t.Fatalf("writing present file: %v", err) + } + + refs := []PresentFileRef{ + {ID: 1, Path: present}, + {ID: 2, Path: filepath.Join(dir, "gone.mkv")}, + {ID: 3, Path: filepath.Join(dir, "also-gone.mkv")}, + } + + vanished, err := statVanished(context.Background(), refs) + if err != nil { + t.Fatalf("statVanished: %v", err) + } + slices.Sort(vanished) + if !slices.Equal(vanished, []int{2, 3}) { + t.Fatalf("vanished = %v, want [2 3]", vanished) + } +} + +// A path whose parent directory is unreadable stats with EACCES, not ENOENT. +// The sweep must leave those files alone: unknown state is not removal. +func TestStatVanishedIgnoresNonNotExistErrors(t *testing.T) { + t.Parallel() + + if os.Geteuid() == 0 { + t.Skip("running as root: permission bits do not deny access") + } + + dir := t.TempDir() + locked := filepath.Join(dir, "locked") + if err := os.Mkdir(locked, 0o755); err != nil { + t.Fatalf("creating locked dir: %v", err) + } + target := filepath.Join(locked, "movie.mkv") + if err := os.WriteFile(target, []byte("x"), 0o644); err != nil { + t.Fatalf("writing target: %v", err) + } + if err := os.Chmod(locked, 0o000); err != nil { + t.Fatalf("chmod: %v", err) + } + t.Cleanup(func() { _ = os.Chmod(locked, 0o755) }) + + vanished, err := statVanished(context.Background(), []PresentFileRef{{ID: 1, Path: target}}) + if err != nil { + t.Fatalf("statVanished: %v", err) + } + if len(vanished) != 0 { + t.Fatalf("vanished = %v, want empty: a permission error must not count as removal", vanished) + } +} + +func TestStatVanishedHandlesEmptyInput(t *testing.T) { + t.Parallel() + + vanished, err := statVanished(context.Background(), nil) + if err != nil { + t.Fatalf("statVanished: %v", err) + } + if len(vanished) != 0 { + t.Fatalf("vanished = %v, want empty", vanished) + } +} + +func TestStatVanishedStopsOnCancelledContext(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + refs := make([]PresentFileRef, 0, 64) + for i := range 64 { + refs = append(refs, PresentFileRef{ID: i + 1, Path: filepath.Join(t.TempDir(), "gone.mkv")}) + } + + if _, err := statVanished(ctx, refs); err == nil { + t.Fatal("statVanished returned nil error for a cancelled context") + } +} + +// The bulk-removal backstop is the difference between "a few files were +// deleted" and "the storage went away"; assert the exact boundary so a change +// to either constant is deliberate. +func TestPresenceSweepAbortThreshold(t *testing.T) { + t.Parallel() + + aborts := func(checked, vanished int) bool { + return vanished >= presenceSweepAbortFloor && + float64(vanished) >= float64(checked)*presenceSweepAbortFraction + } + + cases := []struct { + name string + checked, vanished int + want bool + }{ + {"single deletion in a large library", 5000, 1, false}, + {"routine upgrade churn", 5000, 40, false}, + {"small library emptied stays under the floor", 10, 10, false}, + {"half a large library vanishing", 5000, 2500, true}, + {"just under the floor", 40, 19, false}, + {"at the floor and at the fraction", 40, 20, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := aborts(tc.checked, tc.vanished); got != tc.want { + t.Fatalf("aborts(checked=%d, vanished=%d) = %v, want %v", + tc.checked, tc.vanished, got, tc.want) + } + }) + } +} + +func TestVerifyPresenceRequiresConfiguredScanner(t *testing.T) { + t.Parallel() + + var s *Scanner + if _, err := s.VerifyPresence(context.Background(), nil); err == nil { + t.Fatal("VerifyPresence on a nil scanner returned nil error") + } +} diff --git a/internal/taskmanager/tasks/verify_file_presence.go b/internal/taskmanager/tasks/verify_file_presence.go new file mode 100644 index 00000000..8186b873 --- /dev/null +++ b/internal/taskmanager/tasks/verify_file_presence.go @@ -0,0 +1,119 @@ +package tasks + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/scanner" + "github.com/Silo-Server/silo-server/internal/taskmanager" +) + +// defaultVerifyFilePresenceIntervalMs is how often the sweep runs. Fifteen +// minutes is the ceiling on how long a deleted file can keep advertising +// itself as playable when no autoscan source covers the library. +const defaultVerifyFilePresenceIntervalMs int64 = 15 * 60 * 1000 + +// PresenceVerifier stats a library's cataloged files and marks the vanished +// ones missing. Implemented by *scanner.Scanner. +type PresenceVerifier interface { + VerifyPresence(ctx context.Context, folder *models.MediaFolder) (scanner.PresenceSweepResult, error) +} + +// VerifyFilePresenceTask closes the window between a file being deleted on +// disk and the catalog noticing. Catalog reads hide files with missing_since +// set, so marking is all it takes to stop offering a dead file for playback; +// without this the first thing to notice a deletion is a failed play. +type VerifyFilePresenceTask struct { + folderRepo ScanFolderRepository + verifier PresenceVerifier + intervalMs int64 +} + +type verifyFilePresenceResult struct { + Checked int `json:"checked"` + Marked int `json:"marked"` + MembershipsRemoved int `json:"memberships_removed"` + ItemsDeleted int `json:"items_deleted"` + FoldersAborted int `json:"folders_aborted"` + Errors int `json:"errors"` +} + +// NewVerifyFilePresenceTask builds the sweep task. intervalMs is in +// MILLISECONDS; a non-positive value falls back to the default cadence. +func NewVerifyFilePresenceTask(folderRepo ScanFolderRepository, verifier PresenceVerifier, intervalMs int64) *VerifyFilePresenceTask { + if intervalMs <= 0 { + intervalMs = defaultVerifyFilePresenceIntervalMs + } + return &VerifyFilePresenceTask{folderRepo: folderRepo, verifier: verifier, intervalMs: intervalMs} +} + +func (t *VerifyFilePresenceTask) Key() string { return "verify_file_presence" } +func (t *VerifyFilePresenceTask) Name() string { return "Verify Media File Presence" } +func (t *VerifyFilePresenceTask) Description() string { + return "Checks that cataloged media files still exist on disk and hides the ones that were removed, without waiting for a full library scan" +} +func (t *VerifyFilePresenceTask) Category() taskmanager.TaskCategory { + return taskmanager.TaskCategoryLibrary +} +func (t *VerifyFilePresenceTask) IsHidden() bool { return false } + +func (t *VerifyFilePresenceTask) DefaultTriggers() []taskmanager.TriggerConfig { + return []taskmanager.TriggerConfig{ + {Type: taskmanager.TriggerTypeInterval, IntervalMs: t.intervalMs}, + } +} + +func (t *VerifyFilePresenceTask) Execute(ctx context.Context, progress taskmanager.ProgressReporter) error { + if t == nil || t.verifier == nil || t.folderRepo == nil { + progress.Report(100, "File presence verification is not configured") + return nil + } + + folders, err := t.folderRepo.GetEnabled(ctx) + if err != nil { + return fmt.Errorf("listing enabled folders: %w", err) + } + if len(folders) == 0 { + progress.Report(100, "No libraries to verify") + return nil + } + + var summary verifyFilePresenceResult + // Sequential across folders on purpose: VerifyPresence already fans out + // internally, and libraries commonly share one underlying mount. + for i, folder := range folders { + if folder == nil { + continue + } + if err := ctx.Err(); err != nil { + return err + } + + res, verifyErr := t.verifier.VerifyPresence(ctx, folder) + if verifyErr != nil { + slog.ErrorContext(ctx, "presence sweep: failed to verify library", "component", "taskmanager", + "folder_id", folder.ID, "name", folder.Name, "error", verifyErr) + summary.Errors++ + } else { + summary.Checked += res.Checked + summary.Marked += res.Marked + summary.MembershipsRemoved += res.MembershipsRemoved + summary.ItemsDeleted += res.ItemsDeleted + if res.Aborted { + summary.FoldersAborted++ + } + } + + progress.Report(float64(i+1)/float64(len(folders))*100, + fmt.Sprintf("Verified %s (%d/%d libraries)", folder.Name, i+1, len(folders))) + } + + if data, err := json.Marshal(summary); err == nil { + progress.SetResultData(data) + } + progress.Report(100, fmt.Sprintf("Checked %d files, hid %d removed", summary.Checked, summary.Marked)) + return nil +}