feat(scanner): quiet scans, enqueue coalescing, watcher probe, sibling cache
Part 2 of the faceted plan, Phase 5 (steps 5-6): the scan that used to print ~4 lines per unchanged file now prints ONE summary line, and a poll tick landing mid-scan no longer stacks a second full scan. - Quiet scans (§4): the walk loop is silent — inserts, changes, and errors log inside processMediaFile; the duplicate 'Successfully processed' branch (both branches printed it) and the per-file 'processing / already exists / same size, skipping' chatter are gone. Summary carries the trigger and a changed counter: 'Scan of <folders> (trigger: poll): N files, N changed, N new, Xs, N errors'. Triggers threaded through every enqueue site: poll (startBackupScan), watcher (dirty-dirs flush), admin (scan API), initial (startup scan). - Enqueue coalescing (§5): Worker.HasActiveScanForLibrary (JobResult gains additive job_type so only SCAN jobs coalesce — an import sharing the library id never blocks one). enqueueLibraryScan skips silently; the admin scan API answers 409 with the active job id. Enforces the one-scan-per-library invariant the bulk scan-state map leans on. - Watcher capability probe (§4): ONE Add on a root folder; failure disables the watcher with a single startup line (replaces the per-directory warning spam on inotify-broken deployments). The watcher stays an optional latency reducer over scheduled scans. - Per-directory sibling cache (§6): hasSiblingBookFileCached caches the ReadDir per directory for one scan — loose-image manga re-stats thousands of images sharing a handful of directories. Tests: TestHasActiveScanForLibrary (unit — running/pending coalesce, terminal jobs don't, non-scan jobs don't, empty library never, nil worker safe). Full scanner suite green in the container.
This commit is contained in:
1 parent
6750e9e0dc
commit
5245f292ca
5 files changed
+225
-49
No files matched your search
@@ -219,3 +219,21 @@ cover_image_path, has_conflict, format_group, badge`.
|
|||||||
re-extraction (deploy-day rule). Multifile folder/zip books keep the
|
re-extraction (deploy-day rule). Multifile folder/zip books keep the
|
||||||
Phase 3 behavior (their change path does not consume sidecar
|
Phase 3 behavior (their change path does not consume sidecar
|
||||||
extractors yet — gating them would flip books "changed" for nothing).
|
extractors yet — gating them would flip books "changed" for nothing).
|
||||||
|
|
||||||
|
## 2026-10-19 — Scanner Phase 5: quiet scans, coalescing, watcher probe
|
||||||
|
|
||||||
|
- `POST /api/scanner/scan` now answers **409** (with `active_job_id`)
|
||||||
|
when a scan for the same library is already pending/running —
|
||||||
|
coalescing instead of stacking a second full scan. The endpoint was
|
||||||
|
never pinned; no existing response shape changed. Poll ticks and
|
||||||
|
watcher events coalesce silently (no second job enqueued).
|
||||||
|
- Scan jobs carry a `trigger` label (poll/watcher/admin/initial); the
|
||||||
|
per-scan log is now ONE summary line:
|
||||||
|
`Scan of <folders> (trigger: <t>): N files, N changed, N new, Xs, N errors`.
|
||||||
|
Per-file "processing/success/skipping" chatter is gone; inserts,
|
||||||
|
changes, and errors still log individually.
|
||||||
|
- `JobResult` gains additive `job_type` (coalescing needs it; the
|
||||||
|
scan-progress surfaces tolerate unknown keys).
|
||||||
|
- Watcher: a failed capability probe (one `Add` on a root folder)
|
||||||
|
disables the watcher with ONE startup line instead of per-directory
|
||||||
|
warning spam.
|
||||||
@@ -84,6 +84,23 @@ func (h *Handler) ScanLibrary(c *echo.Context) error {
|
|||||||
return c.JSON(http.StatusBadRequest, map[string]string{"error": "either library_id or folder_paths required for scanning"})
|
return c.JSON(http.StatusBadRequest, map[string]string{"error": "either library_id or folder_paths required for scanning"})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Coalescing (plan Part 2 §5): a scan for this library is already
|
||||||
|
// pending or running — 409 with the active job so the UI can show
|
||||||
|
// progress instead of stacking a second full scan.
|
||||||
|
if req.LibraryID != "" && h.worker.HasActiveScanForLibrary(req.LibraryID) {
|
||||||
|
var activeJobID string
|
||||||
|
for _, active := range h.worker.GetActiveScanJobs() {
|
||||||
|
if active.JobType == services.JobTypeScan && active.LibraryID == req.LibraryID {
|
||||||
|
activeJobID = active.JobID
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return c.JSON(http.StatusConflict, map[string]string{
|
||||||
|
"error": "a scan for this library is already pending or running",
|
||||||
|
"active_job_id": activeJobID,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
jobID := uuid.New().String()
|
jobID := uuid.New().String()
|
||||||
|
|
||||||
job := &services.Job{
|
job := &services.Job{
|
||||||
@@ -96,6 +113,7 @@ func (h *Handler) ScanLibrary(c *echo.Context) error {
|
|||||||
"admin_id": userUUID.String(),
|
"admin_id": userUUID.String(),
|
||||||
"db": h.db,
|
"db": h.db,
|
||||||
"force": req.Force,
|
"force": req.Force,
|
||||||
|
"trigger": "admin",
|
||||||
},
|
},
|
||||||
Status: services.JobStatusPending,
|
Status: services.JobStatusPending,
|
||||||
Context: h.ctx,
|
Context: h.ctx,
|
||||||
|
|||||||
@@ -168,6 +168,15 @@ type MediaScanner struct {
|
|||||||
scanState map[string]*scanEntry
|
scanState map[string]*scanEntry
|
||||||
scanStateMu sync.RWMutex
|
scanStateMu sync.RWMutex
|
||||||
folderLibrary map[string]pgtype.UUID
|
folderLibrary map[string]pgtype.UUID
|
||||||
|
|
||||||
|
// Summary + cache state (faceted plan Part 2 §4-6): scanTrigger
|
||||||
|
// labels the per-scan summary line (poll/watcher/admin/initial);
|
||||||
|
// changedItems counts re-processes for it; siblingBookCache caches
|
||||||
|
// hasSiblingBookFile's ReadDir per directory for one scan (loose-
|
||||||
|
// image manga re-stats thousands of images in a handful of dirs).
|
||||||
|
scanTrigger string
|
||||||
|
changedItems int
|
||||||
|
siblingBookCache map[string]bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// scanEntry is one media item's scanner-owned facts, bulk-loaded per
|
// scanEntry is one media item's scanner-owned facts, bulk-loaded per
|
||||||
@@ -435,6 +444,18 @@ func (s *MediaScanner) SetForce(force bool) {
|
|||||||
s.forceRescan = force
|
s.forceRescan = force
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SetTrigger labels the scan's summary line (plan Part 2 §4): poll,
|
||||||
|
// watcher, admin, initial.
|
||||||
|
func (s *MediaScanner) SetTrigger(trigger string) {
|
||||||
|
s.scanTrigger = trigger
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetChangedItems reports how many existing items re-processed this run
|
||||||
|
// (size change, sidecar change, or force rescan).
|
||||||
|
func (s *MediaScanner) GetChangedItems() int {
|
||||||
|
return s.changedItems
|
||||||
|
}
|
||||||
|
|
||||||
func (s *MediaScanner) GetStats() (int, int, int) {
|
func (s *MediaScanner) GetStats() (int, int, int) {
|
||||||
return s.totalFiles, s.newItems, s.errors
|
return s.totalFiles, s.newItems, s.errors
|
||||||
}
|
}
|
||||||
@@ -500,30 +521,35 @@ func (s *MediaScanner) SetFolders(folders []string, watch bool) error {
|
|||||||
// Add all folders and their subdirectories to the watcher (like Audiobookshelf).
|
// Add all folders and their subdirectories to the watcher (like Audiobookshelf).
|
||||||
// Only when watching; scan jobs (watch=false) skip this entirely.
|
// Only when watching; scan jobs (watch=false) skip this entirely.
|
||||||
if s.watcher != nil {
|
if s.watcher != nil {
|
||||||
watchCount := 0
|
// Capability probe (plan Part 2 §4): ONE Add on a root folder
|
||||||
for _, folder := range folders {
|
// before committing. inotify-broken deployments (watch limits,
|
||||||
if err := s.watcher.Add(folder); err != nil {
|
// network mounts) used to spam one warning per directory; on
|
||||||
fmt.Printf("[WATCHER] Warning: failed to watch root folder %s: %v\n", folder, err)
|
// probe failure the watcher is disabled with a single line — it
|
||||||
} else {
|
// stays an optional latency reducer over the scheduled scans.
|
||||||
watchCount++
|
if err := s.watcher.Add(folders[0]); err != nil {
|
||||||
}
|
fmt.Printf("[WATCHER] Disabled: cannot watch %s (%v) — scheduled scans unaffected, only change-triggered latency is lost\n", folders[0], err)
|
||||||
filepath.WalkDir(folder, func(path string, d fs.DirEntry, err error) error {
|
_ = s.watcher.Close()
|
||||||
if err != nil {
|
s.watcher = nil
|
||||||
return err
|
} else {
|
||||||
}
|
watchCount := 1
|
||||||
if !d.IsDir() || path == folder {
|
for _, folder := range folders {
|
||||||
|
_ = s.watcher.Add(folder) // probe proved the capability; failures here are counted, not spammed
|
||||||
|
filepath.WalkDir(folder, func(path string, d fs.DirEntry, err error) error {
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if !d.IsDir() || path == folder {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if err := s.watcher.Add(path); err == nil {
|
||||||
|
watchCount++
|
||||||
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
})
|
||||||
if err := s.watcher.Add(path); err != nil {
|
}
|
||||||
fmt.Printf("[WATCHER] Warning: failed to watch subdirectory %s: %v\n", path, err)
|
|
||||||
} else {
|
|
||||||
watchCount++
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
fmt.Printf("[WATCHER] Now watching %d directories across %d root folders\n", watchCount, len(folders))
|
fmt.Printf("[WATCHER] Now watching %d directories across %d root folders\n", watchCount, len(folders))
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
fmt.Printf("[SCANNER] Configured %d root folders (watch mode disabled, no inotify watcher)\n", len(folders))
|
fmt.Printf("[SCANNER] Configured %d root folders (watch mode disabled, no inotify watcher)\n", len(folders))
|
||||||
}
|
}
|
||||||
@@ -531,7 +557,7 @@ func (s *MediaScanner) SetFolders(folders []string, watch bool) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *MediaScanner) enqueueLibraryScan(rootFolder string) {
|
func (s *MediaScanner) enqueueLibraryScan(rootFolder, trigger string) {
|
||||||
if s.db == nil {
|
if s.db == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -569,6 +595,14 @@ func (s *MediaScanner) enqueueLibraryScan(rootFolder string) {
|
|||||||
|
|
||||||
libraryIDStr := uuid.UUID(libRow.LibraryID.Bytes).String()
|
libraryIDStr := uuid.UUID(libRow.LibraryID.Bytes).String()
|
||||||
|
|
||||||
|
// Coalescing (plan Part 2 §5): a pending/running scan for the SAME
|
||||||
|
// library makes this enqueue a no-op — a poll tick that lands while
|
||||||
|
// a scan is still running must not stack a second full scan, and
|
||||||
|
// the bulk scan-state map leans on one-scan-per-library.
|
||||||
|
if WorkerInstance.HasActiveScanForLibrary(libraryIDStr) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
job := &Job{
|
job := &Job{
|
||||||
ID: uuid.New().String(),
|
ID: uuid.New().String(),
|
||||||
Type: JobTypeScan,
|
Type: JobTypeScan,
|
||||||
@@ -581,12 +615,12 @@ func (s *MediaScanner) enqueueLibraryScan(rootFolder string) {
|
|||||||
"admin_id": adminIDStr,
|
"admin_id": adminIDStr,
|
||||||
"db": s.db,
|
"db": s.db,
|
||||||
"force": false,
|
"force": false,
|
||||||
|
"trigger": trigger,
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
if WorkerInstance != nil {
|
if WorkerInstance != nil {
|
||||||
WorkerInstance.Enqueue(job)
|
WorkerInstance.Enqueue(job)
|
||||||
fmt.Printf("[MTIME-POLL] Enqueued library scan for %s (library: %s)\n", rootFolder, libraryIDStr)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -606,11 +640,13 @@ func (s *MediaScanner) ScanFolders(ctx context.Context) error {
|
|||||||
s.totalFiles = 0
|
s.totalFiles = 0
|
||||||
s.newItems = 0
|
s.newItems = 0
|
||||||
s.errors = 0
|
s.errors = 0
|
||||||
|
s.changedItems = 0
|
||||||
s.audioBookRoots = make(map[string]bool)
|
s.audioBookRoots = make(map[string]bool)
|
||||||
// Bulk scan state: per-run reset; one load per distinct library.
|
// Bulk scan state: per-run reset; one load per distinct library.
|
||||||
s.scanStateMu.Lock()
|
s.scanStateMu.Lock()
|
||||||
s.scanState = make(map[string]*scanEntry)
|
s.scanState = make(map[string]*scanEntry)
|
||||||
s.folderLibrary = make(map[string]pgtype.UUID)
|
s.folderLibrary = make(map[string]pgtype.UUID)
|
||||||
|
s.siblingBookCache = make(map[string]bool)
|
||||||
s.scanStateMu.Unlock()
|
s.scanStateMu.Unlock()
|
||||||
loadedLibraries := make(map[string]bool)
|
loadedLibraries := make(map[string]bool)
|
||||||
|
|
||||||
@@ -631,8 +667,8 @@ func (s *MediaScanner) ScanFolders(ctx context.Context) error {
|
|||||||
|
|
||||||
fmt.Printf("Starting scan of %d folders: %v (%d files to scan)\n", len(s.folders), s.folders, s.totalFiles)
|
fmt.Printf("Starting scan of %d folders: %v (%d files to scan)\n", len(s.folders), s.folders, s.totalFiles)
|
||||||
|
|
||||||
|
scanStart := time.Now()
|
||||||
processedFiles := 0
|
processedFiles := 0
|
||||||
mediaFiles := 0
|
|
||||||
|
|
||||||
for _, folder := range s.folders {
|
for _, folder := range s.folders {
|
||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil {
|
||||||
@@ -686,7 +722,6 @@ func (s *MediaScanner) ScanFolders(ctx context.Context) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if s.isScannableFile(path) {
|
if s.isScannableFile(path) {
|
||||||
mediaFiles++
|
|
||||||
processedFiles++
|
processedFiles++
|
||||||
|
|
||||||
if processedFiles%10 == 0 && s.totalFiles > 0 {
|
if processedFiles%10 == 0 && s.totalFiles > 0 {
|
||||||
@@ -709,14 +744,13 @@ func (s *MediaScanner) ScanFolders(ctx context.Context) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
wasNew, err := s.processMediaFile(ctx, path)
|
// Quiet scans (plan Part 2 §4): inserts, changes and
|
||||||
if err != nil {
|
// errors log inside processMediaFile — the walk itself
|
||||||
|
// stays silent so an unchanged library costs ~1 summary
|
||||||
|
// line, not one per file.
|
||||||
|
if _, err := s.processMediaFile(ctx, path); err != nil {
|
||||||
fmt.Printf("Error processing media file %s: %v\n", path, err)
|
fmt.Printf("Error processing media file %s: %v\n", path, err)
|
||||||
s.errors++
|
s.errors++
|
||||||
} else if wasNew {
|
|
||||||
fmt.Printf("Successfully processed media file: %s\n", path)
|
|
||||||
} else {
|
|
||||||
fmt.Printf("Successfully processed media file: %s\n", path)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -728,8 +762,13 @@ func (s *MediaScanner) ScanFolders(ctx context.Context) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fmt.Printf("Scan completed: %d total files scanned, %d media files found, %d new items, %d errors\n",
|
elapsed := time.Since(scanStart)
|
||||||
processedFiles, mediaFiles, s.newItems, s.errors)
|
foldersLabel := strings.Join(s.folders, ", ")
|
||||||
|
if s.scanTrigger == "" {
|
||||||
|
s.scanTrigger = "unknown"
|
||||||
|
}
|
||||||
|
fmt.Printf("Scan of %s (trigger: %s): %d files, %d changed, %d new, %.1fs, %d errors\n",
|
||||||
|
foldersLabel, s.scanTrigger, processedFiles, s.changedItems, s.newItems, elapsed.Seconds(), s.errors)
|
||||||
|
|
||||||
// A cancelled scan must not run the archive lifecycle pass below:
|
// A cancelled scan must not run the archive lifecycle pass below:
|
||||||
// it never visited every folder, so unvisited files would be
|
// it never visited every folder, so unvisited files would be
|
||||||
@@ -958,9 +997,29 @@ func hasSiblingBookFile(dir string) bool {
|
|||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *MediaScanner) processMediaFile(ctx context.Context, path string) (bool, error) {
|
// hasSiblingBookFileCached is the per-directory cached variant (plan
|
||||||
fmt.Printf("Processing media file: %s\n", path)
|
// Part 2 §6): loose-image manga stats thousands of images that share a
|
||||||
|
// handful of directories — one ReadDir per directory per scan, not per
|
||||||
|
// image. Cache is reset with the scan run; a nil cache (watcher path)
|
||||||
|
// computes uncached.
|
||||||
|
func (s *MediaScanner) hasSiblingBookFileCached(dir string) bool {
|
||||||
|
s.scanStateMu.RLock()
|
||||||
|
cached, ok := s.siblingBookCache[dir]
|
||||||
|
cache := s.siblingBookCache
|
||||||
|
s.scanStateMu.RUnlock()
|
||||||
|
if ok {
|
||||||
|
return cached
|
||||||
|
}
|
||||||
|
result := hasSiblingBookFile(dir)
|
||||||
|
if cache != nil {
|
||||||
|
s.scanStateMu.Lock()
|
||||||
|
cache[dir] = result
|
||||||
|
s.scanStateMu.Unlock()
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *MediaScanner) processMediaFile(ctx context.Context, path string) (bool, error) {
|
||||||
// Get file info
|
// Get file info
|
||||||
info, err := os.Stat(path)
|
info, err := os.Stat(path)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -968,7 +1027,7 @@ func (s *MediaScanner) processMediaFile(ctx context.Context, path string) (bool,
|
|||||||
return false, fmt.Errorf("failed to get file info: %v", err)
|
return false, fmt.Errorf("failed to get file info: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if isImageFile(path) && hasSiblingBookFile(filepath.Dir(path)) {
|
if isImageFile(path) && s.hasSiblingBookFileCached(filepath.Dir(path)) {
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1025,7 +1084,8 @@ func (s *MediaScanner) processMediaFile(ctx context.Context, path string) (bool,
|
|||||||
return false, fmt.Errorf("failed to check if media item exists: %v", err)
|
return false, fmt.Errorf("failed to check if media item exists: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
fmt.Printf("Media item does not exist in database, creating new entry: %s\n", path)
|
// Quiet path (plan Part 2 §4): map miss in bulk mode means NEW —
|
||||||
|
// proceed straight to extraction + create.
|
||||||
|
|
||||||
// Extract metadata from file first
|
// Extract metadata from file first
|
||||||
metadata, err := s.extractMetadata(path)
|
metadata, err := s.extractMetadata(path)
|
||||||
@@ -1300,6 +1360,7 @@ func (s *MediaScanner) processMediaFile(ctx context.Context, path string) (bool,
|
|||||||
}
|
}
|
||||||
|
|
||||||
s.newItems++
|
s.newItems++
|
||||||
|
fmt.Printf("[SCAN] Added new media item: %s\n", path)
|
||||||
|
|
||||||
// A2c §2.4: obvious playlist sidecar series become curated collections.
|
// A2c §2.4: obvious playlist sidecar series become curated collections.
|
||||||
s.syncPlaylistCollectionsAfterCreate(ctx, createdItem.ID, metadata.PlaylistSeries)
|
s.syncPlaylistCollectionsAfterCreate(ctx, createdItem.ID, metadata.PlaylistSeries)
|
||||||
@@ -1330,12 +1391,11 @@ func (s *MediaScanner) processMediaFile(ctx context.Context, path string) (bool,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// processExistingRow is the nil-map (per-file lookup) existing-item
|
// processExistingRow is the nil-map (per-file lookup) existing-item
|
||||||
// branch, preserved verbatim from the pre-bulk-load behavior so the
|
// branch, preserved from the pre-bulk-load behavior so the watcher's
|
||||||
// watcher's scanDirectory path and any bulk-load failure keep working
|
// scanDirectory path and any bulk-load failure keep working exactly as
|
||||||
// exactly as before.
|
// before. Quiet-scan rules (plan Part 2 §4): only changes and errors
|
||||||
|
// log — unchanged skips are silent.
|
||||||
func (s *MediaScanner) processExistingRow(ctx context.Context, existingItem database.MediaItems, libraryID pgtype.UUID, path string, info os.FileInfo) (bool, error) {
|
func (s *MediaScanner) processExistingRow(ctx context.Context, existingItem database.MediaItems, libraryID pgtype.UUID, path string, info os.FileInfo) (bool, error) {
|
||||||
fmt.Printf("Media item already exists in database: %s (size: %d vs %d)\n", path, existingItem.FileSize.Int64, info.Size())
|
|
||||||
|
|
||||||
// The file is back on disk: lift any archive/missing state so the
|
// The file is back on disk: lift any archive/missing state so the
|
||||||
// item reappears in libraries and future missing scans start fresh.
|
// item reappears in libraries and future missing scans start fresh.
|
||||||
if existingItem.ArchivedAt.Valid || existingItem.MissingScanCount > 0 {
|
if existingItem.ArchivedAt.Valid || existingItem.MissingScanCount > 0 {
|
||||||
@@ -1349,7 +1409,6 @@ func (s *MediaScanner) processExistingRow(ctx context.Context, existingItem data
|
|||||||
|
|
||||||
// If force rescan is enabled, always re-process
|
// If force rescan is enabled, always re-process
|
||||||
if s.forceRescan {
|
if s.forceRescan {
|
||||||
fmt.Printf("Force rescan enabled, updating existing media item: %s\n", path)
|
|
||||||
// Use UPDATE instead of DELETE+INSERT to preserve created_at
|
// Use UPDATE instead of DELETE+INSERT to preserve created_at
|
||||||
if err := s.updateMediaItem(ctx, existingItem, path); err != nil {
|
if err := s.updateMediaItem(ctx, existingItem, path); err != nil {
|
||||||
fmt.Printf("Warning: failed to update existing media item: %v\n", err)
|
fmt.Printf("Warning: failed to update existing media item: %v\n", err)
|
||||||
@@ -1357,6 +1416,7 @@ func (s *MediaScanner) processExistingRow(ctx context.Context, existingItem data
|
|||||||
// Recompute hash identifiers too - a force rescan is the admin's
|
// Recompute hash identifiers too - a force rescan is the admin's
|
||||||
// backfill tool and must refresh stale or missing hashes.
|
// backfill tool and must refresh stale or missing hashes.
|
||||||
s.recomputeHashInfo(ctx, existingItem.ID, libraryID, path)
|
s.recomputeHashInfo(ctx, existingItem.ID, libraryID, path)
|
||||||
|
s.changedItems++
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
// Normal behavior: check if file has changed (by size)
|
// Normal behavior: check if file has changed (by size)
|
||||||
@@ -1365,6 +1425,7 @@ func (s *MediaScanner) processExistingRow(ctx context.Context, existingItem data
|
|||||||
_ = s.updateMediaItem(ctx, existingItem, path)
|
_ = s.updateMediaItem(ctx, existingItem, path)
|
||||||
// The bytes changed, so any stored hash is stale.
|
// The bytes changed, so any stored hash is stale.
|
||||||
s.recomputeHashInfo(ctx, existingItem.ID, libraryID, path)
|
s.recomputeHashInfo(ctx, existingItem.ID, libraryID, path)
|
||||||
|
s.changedItems++
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
// Self-heal items imported before hashing existed: even an unchanged
|
// Self-heal items imported before hashing existed: even an unchanged
|
||||||
@@ -1372,7 +1433,6 @@ func (s *MediaScanner) processExistingRow(ctx context.Context, existingItem data
|
|||||||
if !existingItem.FileSha256.Valid || existingItem.FileSha256.String == "" {
|
if !existingItem.FileSha256.Valid || existingItem.FileSha256.String == "" {
|
||||||
s.recomputeHashInfo(ctx, existingItem.ID, libraryID, path)
|
s.recomputeHashInfo(ctx, existingItem.ID, libraryID, path)
|
||||||
}
|
}
|
||||||
fmt.Printf("Media item already exists with same size, skipping: %s\n", path)
|
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1397,7 +1457,6 @@ func (s *MediaScanner) processExistingViaScanState(ctx context.Context, entry *s
|
|||||||
}
|
}
|
||||||
|
|
||||||
if s.forceRescan {
|
if s.forceRescan {
|
||||||
fmt.Printf("Force rescan enabled, updating existing media item: %s\n", path)
|
|
||||||
existingItem, err := s.getMediaItemByFilePath(ctx, path, libraryID)
|
existingItem, err := s.getMediaItemByFilePath(ctx, path, libraryID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, fmt.Errorf("force rescan re-read failed: %v", err)
|
return false, fmt.Errorf("force rescan re-read failed: %v", err)
|
||||||
@@ -1408,6 +1467,7 @@ func (s *MediaScanner) processExistingViaScanState(ctx context.Context, entry *s
|
|||||||
// Recompute hash identifiers too - a force rescan is the admin's
|
// Recompute hash identifiers too - a force rescan is the admin's
|
||||||
// backfill tool and must refresh stale or missing hashes.
|
// backfill tool and must refresh stale or missing hashes.
|
||||||
s.recomputeHashInfo(ctx, entry.ID, libraryID, path)
|
s.recomputeHashInfo(ctx, entry.ID, libraryID, path)
|
||||||
|
s.changedItems++
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1424,6 +1484,7 @@ func (s *MediaScanner) processExistingViaScanState(ctx context.Context, entry *s
|
|||||||
s.recomputeHashInfo(ctx, entry.ID, libraryID, path)
|
s.recomputeHashInfo(ctx, entry.ID, libraryID, path)
|
||||||
entry.FileSize = info.Size()
|
entry.FileSize = info.Size()
|
||||||
s.scanStateUpsert(libraryID, entry)
|
s.scanStateUpsert(libraryID, entry)
|
||||||
|
s.changedItems++
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1455,6 +1516,7 @@ func (s *MediaScanner) processExistingViaScanState(ctx context.Context, entry *s
|
|||||||
// sidecars, not the stale pre-edit fingerprint.
|
// sidecars, not the stale pre-edit fingerprint.
|
||||||
dir := bookSidecarDir(path)
|
dir := bookSidecarDir(path)
|
||||||
s.storeManifest(ctx, libraryID, entry, hashSidecarManifest(path, dir, statSidecarManifest(path, dir)))
|
s.storeManifest(ctx, libraryID, entry, hashSidecarManifest(path, dir, statSidecarManifest(path, dir)))
|
||||||
|
s.changedItems++
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -4253,7 +4315,7 @@ func (s *MediaScanner) processDirtyDirectories(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
for rootFolder := range affectedRoots {
|
for rootFolder := range affectedRoots {
|
||||||
s.enqueueLibraryScan(rootFolder)
|
s.enqueueLibraryScan(rootFolder, "watcher")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -4424,6 +4486,7 @@ func (s *MediaScanner) performInitialScan(ctx context.Context) {
|
|||||||
"admin_id": adminIDStr,
|
"admin_id": adminIDStr,
|
||||||
"db": s.db,
|
"db": s.db,
|
||||||
"force": false,
|
"force": false,
|
||||||
|
"trigger": "initial",
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -4503,7 +4566,7 @@ func (s *MediaScanner) startBackupScan(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
fmt.Printf("[BACKUP-SCAN] Running periodic full scan (interval: %v)...\n", interval)
|
fmt.Printf("[BACKUP-SCAN] Running periodic full scan (interval: %v)...\n", interval)
|
||||||
for _, folder := range s.folders {
|
for _, folder := range s.folders {
|
||||||
s.enqueueLibraryScan(folder)
|
s.enqueueLibraryScan(folder, "poll")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -72,6 +72,7 @@ func (w *Worker) recordPending(job *Job) {
|
|||||||
JobID: job.ID,
|
JobID: job.ID,
|
||||||
Status: JobStatusPending,
|
Status: JobStatusPending,
|
||||||
LibraryID: libID,
|
LibraryID: libID,
|
||||||
|
JobType: job.Type,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -104,7 +105,11 @@ type JobResult struct {
|
|||||||
// LibraryID carries the scan job's library (from Params) so clients
|
// LibraryID carries the scan job's library (from Params) so clients
|
||||||
// can label progress rows after a page refresh, when the original
|
// can label progress rows after a page refresh, when the original
|
||||||
// start-request context is gone.
|
// start-request context is gone.
|
||||||
LibraryID string `json:"library_id,omitempty"`
|
LibraryID string `json:"library_id,omitempty"`
|
||||||
|
// JobType distinguishes scan jobs in the active set — the enqueue
|
||||||
|
// coalescing check (HasActiveScanForLibrary) must only coalesce
|
||||||
|
// scans, never imports/converts sharing a library id.
|
||||||
|
JobType JobType `json:"job_type,omitempty"`
|
||||||
Error string `json:"error"`
|
Error string `json:"error"`
|
||||||
Result interface{} `json:"result"`
|
Result interface{} `json:"result"`
|
||||||
Progress float64 `json:"progress"`
|
Progress float64 `json:"progress"`
|
||||||
@@ -286,6 +291,7 @@ func (w *Worker) processJob(job *Job) {
|
|||||||
JobID: job.ID,
|
JobID: job.ID,
|
||||||
Status: JobStatusRunning,
|
Status: JobStatusRunning,
|
||||||
LibraryID: libID,
|
LibraryID: libID,
|
||||||
|
JobType: job.Type,
|
||||||
}
|
}
|
||||||
|
|
||||||
// Derive a per-job context so cancelling this job (CancelJob) stops
|
// Derive a per-job context so cancelling this job (CancelJob) stops
|
||||||
@@ -486,6 +492,12 @@ func (w *Worker) processScanJob(job *Job) (interface{}, error) {
|
|||||||
scanner.SetAdminID(adminUUID)
|
scanner.SetAdminID(adminUUID)
|
||||||
scanner.SetForce(force)
|
scanner.SetForce(force)
|
||||||
|
|
||||||
|
// Trigger label (plan Part 2 §4): poll / watcher / admin / initial —
|
||||||
|
// surfaces in the per-scan summary line.
|
||||||
|
if trigger, ok := job.Params["trigger"].(string); ok {
|
||||||
|
scanner.SetTrigger(trigger)
|
||||||
|
}
|
||||||
|
|
||||||
var libraryUUID pgtype.UUID
|
var libraryUUID pgtype.UUID
|
||||||
if err := libraryUUID.Scan(libraryID); err != nil {
|
if err := libraryUUID.Scan(libraryID); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -1073,6 +1085,28 @@ func (w *Worker) GetActiveScanJobs() []JobResult {
|
|||||||
return active
|
return active
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// HasActiveScanForLibrary reports whether a pending or running SCAN job
|
||||||
|
// already targets the library (faceted plan Part 2 §5): coalescing
|
||||||
|
// stops poll ticks from stacking full scans when a scan outlives its
|
||||||
|
// interval, and enforces the one-scan-per-library invariant the bulk
|
||||||
|
// scan-state map leans on.
|
||||||
|
func (w *Worker) HasActiveScanForLibrary(libraryID string) bool {
|
||||||
|
if w == nil || libraryID == "" {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
w.mu.RLock()
|
||||||
|
defer w.mu.RUnlock()
|
||||||
|
for _, result := range w.results {
|
||||||
|
if result.JobType != JobTypeScan || result.LibraryID != libraryID {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if result.Status == JobStatusRunning || result.Status == JobStatusPending {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
// CancelJob cancels a pending or running job by id. Pending jobs are
|
// CancelJob cancels a pending or running job by id. Pending jobs are
|
||||||
// marked cancelled so processJob skips them at dequeue (no context
|
// marked cancelled so processJob skips them at dequeue (no context
|
||||||
// exists yet); running jobs have their per-job context cancelled — the
|
// exists yet); running jobs have their per-job context cancelled — the
|
||||||
|
|||||||
@@ -58,3 +58,46 @@ func TestGetActiveScanJobsEmpty(t *testing.T) {
|
|||||||
t.Fatalf("expected no active jobs, got %d", len(jobs))
|
t.Fatalf("expected no active jobs, got %d", len(jobs))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// HasActiveScanForLibrary is the enqueue-coalescing predicate (plan
|
||||||
|
// Part 2 §5): a pending/running SCAN for the same library makes the
|
||||||
|
// next poll tick / watcher event a no-op. Only scans coalesce — an
|
||||||
|
// import or convert sharing the library id must not block one.
|
||||||
|
func TestHasActiveScanForLibrary(t *testing.T) {
|
||||||
|
w := NewWorker(1, nil)
|
||||||
|
|
||||||
|
seed := map[string]*JobResult{
|
||||||
|
"running-scan": {JobID: "running-scan", Status: JobStatusRunning, LibraryID: "lib-1", JobType: JobTypeScan},
|
||||||
|
"pending-scan": {JobID: "pending-scan", Status: JobStatusPending, LibraryID: "lib-2", JobType: JobTypeScan},
|
||||||
|
"done-scan": {JobID: "done-scan", Status: JobStatusCompleted, LibraryID: "lib-3", JobType: JobTypeScan},
|
||||||
|
"failed-scan": {JobID: "failed-scan", Status: JobStatusFailed, LibraryID: "lib-4", JobType: JobTypeScan},
|
||||||
|
"import-lib1": {JobID: "import-lib1", Status: JobStatusRunning, LibraryID: "lib-1", JobType: JobTypeImport},
|
||||||
|
"no-library": {JobID: "no-library", Status: JobStatusRunning, JobType: JobTypeScan},
|
||||||
|
}
|
||||||
|
w.mu.Lock()
|
||||||
|
w.results = seed
|
||||||
|
w.mu.Unlock()
|
||||||
|
|
||||||
|
cases := []struct {
|
||||||
|
library string
|
||||||
|
want bool
|
||||||
|
}{
|
||||||
|
{"lib-1", true}, // running scan
|
||||||
|
{"lib-2", true}, // pending scan
|
||||||
|
{"lib-3", false}, // completed = terminal
|
||||||
|
{"lib-4", false}, // failed = terminal
|
||||||
|
{"lib-9", false}, // unknown library
|
||||||
|
{"", false}, // jobs without a library never coalesce
|
||||||
|
}
|
||||||
|
for _, c := range cases {
|
||||||
|
if got := w.HasActiveScanForLibrary(c.library); got != c.want {
|
||||||
|
t.Errorf("HasActiveScanForLibrary(%q) = %v, want %v", c.library, got, c.want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A nil worker (tests, early startup) must not panic.
|
||||||
|
var nilWorker *Worker
|
||||||
|
if nilWorker.HasActiveScanForLibrary("lib-1") {
|
||||||
|
t.Error("nil worker must report no active scans")
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in new issue
Block a user