Files
bookhoard/internal/handlers/scanner.go
T
john-okeefe 59d5de3607 fix(scanner): eliminate fsnotify watcher leak and harden worker against panics
The bookhoard container crashed with 'panic: Failed to create file
watcher: too many open files' (media_scanner.go) after running for a few
hours, preceded by floods of 'no space left on device' from watcher.Add.

Root cause: every scan job called NewMediaScanner(), which eagerly
created an fsnotify watcher. SetFolders() then walked the entire library
tree and registered one inotify watch per directory (~3,000+ across the
libraries), and ScanFolders() registered them again during its walk. The
worker never called scanner.Close() on these ephemeral per-job scanners,
and the worker loop had no recover(), so:

  1. Leaked watchers accumulated until the kernel inotify watch cap was
     hit (ENOSPC -> 'no space left on device'), then
  2. the process fd limit (ulimit -n 1024) was exhausted, causing
     fsnotify.NewWatcher() to fail with EMFILE, and
  3. NewMediaScanner panicked on that error, taking down the whole
     process (exit code 2). With no restart policy the container stayed
     down.

The scan jobs run frequently (scan_poll_interval), so the leak built up
within hours. Note this was NOT a disk-space issue; df showed plenty free.

Fix:

- media_scanner.go: NewMediaScanner no longer creates a watcher eagerly
  (s.watcher starts nil), which removes the panic site entirely -- there
  is nothing to fail at construction. The watcher is created lazily only
  when needed.

- media_scanner.go: SetFolders gains a [?1049h(B[?7h[?25lEvery 2.0s: boolgaruda-ser8: Fri 31 Jul 2026 10:45:41 AM EDTin 0.002s (127)
sh: line 1: bool: command not found
[?12l[?25h[?1049l
[?1l> parameter. It creates
  and populates a watcher (returning an error instead of panicking) only
  when watch=true; otherwise it skips all watcher.Add calls. ScanFolders
  guards its watcher.Add with a nil check, and the WatchChanges event
  loop exits cleanly when there is no watcher (polling still runs).

- worker.go: the worker() loop now wraps each job in defer/recover() so a
  panicking job is recorded as failed and can never kill the process.

- worker.go: the three ephemeral scan handlers (processScanJob,
  processSetFoldersJob, processDirectoryScanJob) now defer scanner.Close()
  and call SetFolders(..., false), so scan jobs allocate zero watchers and
  zero inotify watches. Any pre-existing leak is also bounded by Close().

- handlers/scanner.go: the long-lived watch-mode scanners (StartScanner
  and StartWatchModeForLibrary) pass watch=true since they actually read
  watcher.Events for live change detection.

- calibre_integration_test.go: updated to the new SetFolders signature
  (watch=false, matching one-off scan usage).

Auto-add is fully preserved: new files are still detected by the periodic
poller (startBackupScan), which is independent of fsnotify and unaffected
by these changes. The watch-mode event loop remains as bonus responsiveness
when inotify is available; through Docker bind mounts where inotify is
unreliable, polling is what catches new books.
2026-07-31 10:45:41 -04:00

336 lines
10 KiB
Go

package handlers
import (
"bookhoard/internal/services"
"context"
"fmt"
"net/http"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
"github.com/labstack/echo/v5"
)
const (
maxPaginationLimit = 1000
)
// StartBackgroundTasks starts the queue processor and cleanup task
// This should be called once for the main handler instance
func (h *Handler) StartBackgroundTasks() {
h.cleanupTaskCancel = h.connManager.StartCleanupTask()
go h.queueProcessor.Start(h.queueCtx)
}
// ScanLibraryRequest represents the request for scanning a library
type ScanLibraryRequest struct {
FolderPaths []string `json:"folder_paths,omitempty"`
LibraryID string `json:"library_id,omitempty"`
Force bool `json:"force,omitempty"`
}
// ScanLibrary handles POST /api/scanner/scan (now runs in background)
func (h *Handler) ScanLibrary(c *echo.Context) error {
var req ScanLibraryRequest
// Check if scan_request is set in context (from library scan route)
if scanReq, ok := c.Get("scan_request").(map[string]interface{}); ok {
if libraryID, ok := scanReq["library_id"].(string); ok {
req.LibraryID = libraryID
}
}
// Bind request body if provided (for direct scanner/scan calls)
if err := c.Bind(&req); err != nil && req.LibraryID == "" {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid request"})
}
// Get user ID from JWT token
userID := c.Get("user_id").(string)
userUUID, err := uuid.Parse(userID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid user id"})
}
var folderPaths []string
// If library_id is provided, fetch folders from library
if req.LibraryID != "" {
libraryUUID, err := uuid.Parse(req.LibraryID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid library id"})
}
// Fetch library folders from database
libraryFolders, err := h.db.GetLibraryFolders(c.Request().Context(), pgtype.UUID{Bytes: libraryUUID, Valid: true})
if err != nil {
return c.JSON(http.StatusNotFound, map[string]string{"error": "library not found or has no folders"})
}
// Extract folder paths
for _, folder := range libraryFolders {
folderPaths = append(folderPaths, folder.FolderPath)
}
if len(folderPaths) == 0 {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "library has no folders configured"})
}
} else if len(req.FolderPaths) > 0 {
// Use folder paths from request
folderPaths = req.FolderPaths
} else {
// Neither library_id nor folder_paths provided
return c.JSON(http.StatusBadRequest, map[string]string{"error": "either library_id or folder_paths required for scanning"})
}
jobID := uuid.New().String()
job := &services.Job{
ID: jobID,
Type: services.JobTypeScan,
UserID: userID,
Params: map[string]interface{}{
"library_id": req.LibraryID,
"folders": folderPaths,
"admin_id": userUUID.String(),
"db": h.db,
"force": req.Force,
},
Status: services.JobStatusPending,
Context: h.ctx,
}
if err := h.worker.EnqueueJob(job); err != nil {
return c.JSON(http.StatusInternalServerError, map[string]string{"error": "failed to enqueue scan job: " + err.Error()})
}
return c.JSON(http.StatusAccepted, map[string]interface{}{
"message": "scan job enqueued",
"job_id": jobID,
"status": "pending",
})
}
// StartScanner handles POST /api/scanner/start
func (h *Handler) StartScanner(c *echo.Context) error {
h.mu.Lock()
defer h.mu.Unlock()
var req ScanLibraryRequest
if err := c.Bind(&req); err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid request"})
}
// Get admin ID from JWT token
userID := c.Get("user_id").(string)
userUUID, err := uuid.Parse(userID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid user id"})
}
// Set the folder paths (watch=true: this long-lived scanner reads events)
if err := h.scanner.SetFolders(req.FolderPaths, true); err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid folder paths: " + err.Error()})
}
// Set the admin ID for media item association
h.scanner.SetAdminID(pgtype.UUID{Bytes: userUUID, Valid: true})
// Start watching for changes
if !h.scanner.GetAutoScanEnabled() {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "auto-scan disabled in settings"})
}
h.scanner.WatchChanges(h.watchModeCtx)
return c.JSON(http.StatusOK, map[string]string{"message": "scanner started"})
}
// StopScanner handles POST /api/scanner/stop
func (h *Handler) StopScanner(c *echo.Context) error {
h.mu.Lock()
defer h.mu.Unlock()
h.watchModeCancel()
h.watchModeCtx, h.watchModeCancel = context.WithCancel(context.Background())
return c.JSON(http.StatusOK, map[string]string{"message": "scanner stopped"})
}
// GetScanStatus handles GET /api/scanner/status/:jobId
func (h *Handler) GetScanStatus(c *echo.Context) error {
jobID := c.Param("jobId")
result, exists := h.worker.GetJobStatus(jobID)
if !exists {
return c.JSON(http.StatusNotFound, map[string]string{"error": "job not found"})
}
return c.JSON(http.StatusOK, map[string]interface{}{
"job_id": result.JobID,
"status": result.Status,
"error": result.Error,
"result": result.Result,
"progress": result.Progress,
"files_scanned": result.FilesScanned,
"new_items": result.NewItems,
"errors": result.Errors,
})
}
// StartWatchModeForLibrary starts watching a specific library's folders
func (h *Handler) StartWatchModeForLibrary(ctx context.Context, libraryID pgtype.UUID, adminID pgtype.UUID) error {
h.mu.Lock()
defer h.mu.Unlock()
libraryIDStr := fmt.Sprintf("%x", libraryID.Bytes)
if h.watchingLibraries[libraryIDStr] {
return fmt.Errorf("already watching library %s", libraryIDStr)
}
folders, err := h.db.GetLibraryFolders(ctx, libraryID)
if err != nil {
return fmt.Errorf("failed to get library folders: %v", err)
}
if len(folders) == 0 {
return fmt.Errorf("no folders configured for library")
}
folderPaths := make([]string, len(folders))
for i, folder := range folders {
folderPaths[i] = folder.FolderPath
}
scanner := services.NewMediaScanner(h.db)
if err := scanner.SetFolders(folderPaths, true); err != nil {
return fmt.Errorf("failed to set scanner folders: %v", err)
}
scanner.SetAdminID(adminID)
scanner.SetLibraryID(libraryID)
scanner.WatchChanges(h.watchModeCtx)
h.watchingLibraries[libraryIDStr] = true
return nil
}
// StopWatchModeForLibrary stops watching a specific library
func (h *Handler) StopWatchModeForLibrary(libraryID pgtype.UUID) error {
h.mu.Lock()
defer h.mu.Unlock()
libraryIDStr := fmt.Sprintf("%x", libraryID.Bytes)
if !h.watchingLibraries[libraryIDStr] {
return fmt.Errorf("not watching library %s", libraryIDStr)
}
delete(h.watchingLibraries, libraryIDStr)
if len(h.watchingLibraries) == 0 {
h.watchModeCancel()
newCtx, newCancel := context.WithCancel(context.Background())
h.watchModeCtx = newCtx
h.watchModeCancel = newCancel
}
return nil
}
// StartWatchMode handles POST /api/scanner/watch/start
func (h *Handler) StartWatchMode(c *echo.Context) error {
userID := c.Get("user_id").(string)
userUUID, err := uuid.Parse(userID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid user id"})
}
var req struct {
LibraryID string `json:"library_id"`
}
if err := c.Bind(&req); err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid request"})
}
if req.LibraryID == "" {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "library_id required"})
}
libraryID, err := uuid.Parse(req.LibraryID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid library id"})
}
if err := h.StartWatchModeForLibrary(c.Request().Context(), pgtype.UUID{Bytes: libraryID, Valid: true}, pgtype.UUID{Bytes: userUUID, Valid: true}); err != nil {
return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()})
}
return c.JSON(http.StatusOK, map[string]interface{}{
"message": "watch mode started for library",
"library_id": req.LibraryID,
})
}
// StopWatchMode handles POST /api/scanner/watch/stop
func (h *Handler) StopWatchMode(c *echo.Context) error {
var req struct {
LibraryID string `json:"library_id"`
}
if err := c.Bind(&req); err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid request"})
}
if req.LibraryID == "" {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "library_id required"})
}
libraryID, err := uuid.Parse(req.LibraryID)
if err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid library id"})
}
if err := h.StopWatchModeForLibrary(pgtype.UUID{Bytes: libraryID, Valid: true}); err != nil {
return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()})
}
return c.JSON(http.StatusOK, map[string]interface{}{
"message": "watch mode stopped for library",
"library_id": req.LibraryID,
})
}
// GetWatchModeStatus handles GET /api/scanner/watch/status
func (h *Handler) GetWatchModeStatus(c *echo.Context) error {
h.mu.Lock()
defer h.mu.Unlock()
watchingLibraries := make([]string, 0, len(h.watchingLibraries))
for libID := range h.watchingLibraries {
watchingLibraries = append(watchingLibraries, libID)
}
return c.JSON(http.StatusOK, map[string]interface{}{
"watching_libraries": watchingLibraries,
"total_watching": len(watchingLibraries),
})
}
// StartWatchModeForAllLibraries starts watching all configured libraries
func (h *Handler) StartWatchModeForAllLibraries(ctx context.Context) error {
libraries, err := h.db.ListLibraries(ctx)
if err != nil {
return fmt.Errorf("failed to list libraries: %v", err)
}
for _, library := range libraries {
libraryIDStr := fmt.Sprintf("%x", library.ID.Bytes)
if err := h.StartWatchModeForLibrary(ctx, library.ID, library.CreatedByAdminID); err != nil {
fmt.Printf("Warning: failed to start watch mode for library %s: %v\n", libraryIDStr, err)
continue
}
fmt.Printf("Started watch mode for library %s (%s)\n", library.Name, libraryIDStr)
}
return nil
}