feat(scanner): detect deleted media files without a full scan
- Add a periodic verify_file_presence task that stats live files per library and marks vanished ones missing, then reconciles memberships - Skip files under unreachable/suspect-empty roots and abort on bulk disappearance, deferring to a full scan - Add FileRepository.ListPresentByFolder and MarkMissingByIDs - Hide files missing from disk in catalog detail playback responses, bounded by a time and count budget
This commit is contained in:
Quick
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user