From 1a552e03f1cf9fdc5af669d1fa2684c397cba34d Mon Sep 17 00:00:00 2001 From: John O'Keefe Date: Thu, 5 Mar 2026 12:55:48 -0500 Subject: [PATCH] docs: Rewrite Phase 0.5 with clean architecture (Phase 0.5 + Phase 1 robustness) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Critical rewrite to fix broken hybrid approach that tried to merge two incompatible systems. PROBLEM WITH PREVIOUS APPROACH: - Tried to use job queue AND direct scanning simultaneously - Created job parameters that didn't match handler expectations - Referenced non-existent activeScans map - Never-initialized worker field in MediaScanner - performInitialScan() bypassed job queue - Like building a car with parts from two different manufacturers CLEAN ARCHITECTURE: - Job queue handles concurrency control ONLY - Scanner handles all scanning logic - Global WorkerInstance provides access (no circular dependency) - Simple scan_mutex for double-protection - Clear separation of concerns KEY CHANGES: 1. MediaScanner struct: - Removed: worker *Worker field (circular dependency) - Removed: activeScans map (too complex) - Fixed: fileStability map[string]*atomic.Bool (was value, now pointer) - Added: scan_mutex sync.Mutex (simple, effective) 2. Job queue integration: - processDirtyDirectories() submits jobs to WorkerInstance - Job parameters: {directory: dirPath, db: s.db} - Added JobTypeDirectoryScan constant - Added processDirectoryScanJob() handler in Worker - performInitialScan() submits jobs (not direct calls) 3. Worker changes: - Added WorkerInstance *Worker global variable - Added Enqueue() method (non-blocking with fallback) - processDirectoryScanJob() creates scanner, calls scanDirectory() PRESERVED FROM PHASE 0.5: - Directory watching with dirty dirs tracking - File stability checks (Audiobookshelf approach) - Smart event merging (Jellyfin approach) - 10-second batch processing - 60-second polling fallback ADDED FROM PHASE 1 ROBUSTNESS: - Job queue for concurrency control - Test isolation - Fixed default values (30s → 60s) RESULT: - No parameter mismatches - No non-existent fields - No memory leaks - Clean separation of concerns - Best of both worlds without the complexity --- COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md | 225 +++++++++++++------- 1 file changed, 151 insertions(+), 74 deletions(-) diff --git a/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md b/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md index 39f4b11..612279d 100644 --- a/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md +++ b/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md @@ -1,7 +1,7 @@ --- -## Phase 0.5: Fix fsnotify Reliability + Job Queue Concurrency Control (3-4 hours) +## Phase 0.5: Fix fsnotify Reliability with Directory Watching + Job Queue Concurrency (3-4 hours) ### Problem Statement Current fsnotify implementation has critical issues: @@ -29,8 +29,8 @@ Current fsnotify implementation has critical issues: - ✅ renameDetection for move operations - ❌ Complex custom implementation -### Solution: Smart Hybrid Approach + Job Queue Concurrency Control -Watch directories (not individual files) with file stability checks + smart event merging + periodic polling fallback + job queue serialization. +### Solution: Clean Phase 0.5 + Phase 1 Robustness +Watch directories (not individual files) with file stability checks + smart event merging + periodic polling fallback. Use job queue ONLY for concurrency control (serialize scans), not for scanning logic. **Key Changes:** 1. Remove per-file event queue (causes overflow) @@ -64,14 +64,18 @@ Watch directories (not individual files) with file stability checks + smart even **Code to ADD:** - `dirtyDirs map[string]time.Time` field - tracks directories pending scan - `dirtyDirsMu sync.RWMutex` field - protects dirtyDirs map -- `fileStability map[string]atomic.Bool` field - tracks files waiting for mtime stabilization (atomic prevents race conditions) +- `fileStability map[string]*atomic.Bool` field - tracks files waiting for mtime stabilization (pointer to atomic, prevents races) - `fileStabilityMu sync.RWMutex` field - protects fileStability map -- `worker *Worker` field - reference to job queue for submitting directory scan jobs +- `scan_mutex sync.Mutex` field - prevents concurrent scans (simple, effective, works with job queue) - `watching atomic.Bool` field - prevents duplicate WatchChanges() calls - `markDirectoryDirty()` function - thread-safe directory marking with smart merging -- `processDirtyDirectories()` function - 10-second batch scanner, submits jobs to worker +- `processDirtyDirectories()` function - 10-second batch scanner, submits jobs to global WorkerInstance - `waitForFileStability()` function - polls mtime until stable (uses atomic.Bool, no race condition) -- `performInitialScan()` function - scans all root folders on startup (submitted as job) +- `performInitialScan()` function - scans all root folders on startup (submitted as jobs, not direct calls) +- `JobTypeDirectoryScan` constant - for directory scanning via job queue +- `processDirectoryScanJob()` function in Worker - handles directory scan jobs +- `WorkerInstance *Worker` global variable - provides access to worker for MediaScanner +- `Enqueue()` method in Worker - non-blocking job submission with safe fallback - `JobTypeSetFolders` constant - for async folder configuration - `processSetFoldersJob()` function - handles folder configuration asynchronously - Test isolation in setupTestServer() - snapshot/restore system_settings @@ -104,9 +108,9 @@ type MediaScanner struct { logger *ScannerLogger dirtyDirs map[string]time.Time // NEW - replaces eventQueue dirtyDirsMu sync.RWMutex // NEW - protects dirtyDirs - fileStability map[string]atomic.Bool // NEW - tracks files waiting (atomic prevents races) + fileStability map[string]*atomic.Bool // NEW - tracks files waiting (atomic prevents races) fileStabilityMu sync.RWMutex // NEW - protects fileStability - worker *Worker // NEW - job queue reference for scan jobs + scan_mutex sync.Mutex // NEW - prevents concurrent directory scans (simple, effective) pollInterval time.Duration watching atomic.Bool // NEW - prevents duplicate calls @@ -141,8 +145,9 @@ func NewMediaScanner(db *database.Queries) *MediaScanner { db: db, watcher: watcher, dirtyDirs: make(map[string]time.Time), - fileStability: make(map[string]atomic.Bool), - pollInterval: 60 * time.Second, // FIXED: Was 30s, now 60s (matches handler/schema) + fileStability: make(map[string]*atomic.Bool), + pollInterval: 60 * time.Second, + watching: atomic.Bool{}, // ... rest of existing initialization } } @@ -238,8 +243,9 @@ const ( ```go const ( - JobTypeScan JobType = "scan" - JobTypeSetFolders JobType = "set_folders" // NEW - async folder configuration + JobTypeScan JobType = "scan" + JobTypeSetFolders JobType = "set_folders" + JobTypeDirectoryScan JobType = "directory_scan" // NEW - Phase 0.5 directory scanning ) ``` @@ -272,6 +278,8 @@ case JobTypeScan: result, err = w.processScanJob(job) case JobTypeSetFolders: result, err = w.processSetFoldersJob(job) +case JobTypeDirectoryScan: + result, err = w.processDirectoryScanJob(job) default: err = fmt.Errorf("unknown job type: %s", job.Type) } @@ -309,6 +317,45 @@ func (w *Worker) processSetFoldersJob(job *Job) (interface{}, error) { "folders": folders, }, nil } + +func (w *Worker) processDirectoryScanJob(job *Job) (interface{}, error) { + // Extract parameters + directoryParam, ok := job.Params["directory"] + if !ok { + return nil, fmt.Errorf("directory parameter required") + } + + directory, ok := directoryParam.(string) + if !ok { + return nil, fmt.Errorf("directory must be a string") + } + + dbParam, ok := job.Params["db"] + if !ok { + return nil, fmt.Errorf("db parameter required") + } + + db, ok := dbParam.(*database.Queries) + if !ok { + return nil, fmt.Errorf("db must be *database.Queries") + } + + // Create temporary scanner instance for this job + scanner := NewMediaScanner(db) + + // Call scanDirectory() directly + // Job queue provides concurrency control - no need for activeScans map + ctx := context.Background() + scanner.scanDirectory(ctx, directory) + + // Return scan results + return map[string]interface{}{ + "message": fmt.Sprintf("Scanned directory: %s", directory), + "totalFiles": scanner.totalFiles, + "newItems": scanner.newItems, + "errors": scanner.errors, + }, nil +} ``` **Why**: @@ -322,6 +369,42 @@ func (w *Worker) processSetFoldersJob(job *Job) (interface{}, error) { --- +### Step 0.5.3.7: Add WorkerInstance Global and Enqueue Method +**File**: `internal/services/worker.go` + +**Location**: After the JobType constants (around line 24-28) + +**Action**: Add global variable and helper method: + +```go +var WorkerInstance *Worker + +func (w *Worker) Enqueue(job *Job) { + select { + case w.jobQueue <- job: + default: + fmt.Printf("Worker queue full, rejecting job: %s\n", job.ID) + } +} +``` + +**Then in main() or worker initialization**: +```go +worker := NewWorker(/* params */) +WorkerInstance = worker +go worker.Start(context.Background()) +``` + +**Why**: +- Provides global access to worker for MediaScanner +- Enqueue() is non-blocking with safe fallback +- No circular dependency between scanner and worker +- Clean separation of concerns + +**Verification**: Run `go build ./internal/services/` to ensure compiles. + +--- + ### Step 0.5.4: Add markDirectoryDirty() Helper with Smart Event Merging **File**: `internal/services/media_scanner.go` @@ -422,32 +505,27 @@ func (s *MediaScanner) processDirtyDirectories(ctx context.Context) { // Process all ready directories in a batch via job queue // Job queue serializes scans - prevents concurrent directory access - if s.worker != nil && len(readyDirs) > 0 { + if len(readyDirs) > 0 { for _, dirPath := range readyDirs { - // Create directory scan job + // Create directory scan job with correct params for processDirectoryScanJob() job := &Job{ ID: uuid.New().String(), - Type: JobTypeScan, + Type: JobTypeDirectoryScan, Params: map[string]interface{}{ - "scan_type": "directory", - "directory": dirPath, - "db": s.db, + "directory": dirPath, + "db": s.db, }, Status: JobStatusPending, } - // Try to enqueue - non-blocking - select { - case s.worker.jobQueue <- job: + // Enqueue via global worker singleton + if WorkerInstance != nil { + WorkerInstance.Enqueue(job) fmt.Printf("Enqueued directory scan job: %s\n", dirPath) - default: - // Worker queue full, skip for now (will be picked up by polling) - fmt.Printf("Worker queue full, skipping directory scan: %s\n", dirPath) + } else { + fmt.Printf("Warning: Worker not initialized, skipping directory scan: %s\n", dirPath) } } - } else if s.worker == nil && len(readyDirs) > 0 { - // Fallback: No worker configured, log warning - fmt.Printf("Warning: No worker configured, %d directories not scanned\n", len(readyDirs)) } } } @@ -459,7 +537,7 @@ reducing redundant scans during bulk operations. Job queue integration ensures: - No concurrent directory scans (serialized by worker pool) - No unbounded goroutine spawn - Natural backpressure when worker queue is full -- Polling fallback will catch missed directories +- Clean separation: job queue handles concurrency, scanner handles logic **Verification**: `go build ./internal/services/` @@ -492,7 +570,7 @@ func (s *MediaScanner) waitForFileStability(filePath string) bool { } // Start tracking with atomic.Bool set to true (checking in progress) - var trackingFlag atomic.Bool + trackingFlag := &atomic.Bool{} trackingFlag.Store(true) s.fileStability[filePath] = trackingFlag s.fileStabilityMu.Unlock() @@ -572,21 +650,10 @@ that are still being copied/downloaded. Polls mtime every 3 seconds until stable ```go func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) { - // FIXED: Prevent concurrent scans of same directory - s.activeScansMu.Lock() - if _, exists := s.activeScans[dirPath]; exists { - s.activeScansMu.Unlock() - return // Already scanning this directory - } - s.activeScans[dirPath] = true - s.activeScansMu.Unlock() - - // FIXED: Ensure cleanup even if panic occurs - defer func() { - s.activeScansMu.Lock() - delete(s.activeScans, dirPath) - s.activeScansMu.Unlock() - }() + // Prevent concurrent scans of ANY directory + // Simple mutex is enough - job queue already serializes by directory + s.scan_mutex.Lock() + defer s.scan_mutex.Unlock() // Find library for this directory var libraryID pgtype.UUID @@ -602,7 +669,7 @@ func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) { } } - // FIXED: Check if libraryID is valid before proceeding + // Check if libraryID is valid before proceeding if !libraryID.Valid { return } @@ -610,13 +677,11 @@ func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) { // Walk directory and process new files filepath.WalkDir(dirPath, func(path string, d os.DirEntry, err error) error { if err != nil { return err } - if d.IsDir() { return filepath.SkipDir } // Skip subdirs (they trigger their own events) + if d.IsDir() { return filepath.SkipDir } if !s.isScannableFile(path) { return nil } // Check if file is stable before processing (Audiobookshelf approach) if !s.waitForFileStability(path) { - // File still being copied, skip for now - // Will be picked up on next poll or directory scan return nil } @@ -641,13 +706,14 @@ func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) { ``` **Why**: Reuses existing scan logic with file stability check (Audiobookshelf approach). -Only processes files that have finished copying/downloading. +Only processes files that have finished copying/downloading. Simple mutex provides +double-protection against concurrent scans, even though job queue already serializes. **Verification**: `go build ./internal/services/` --- -### Step 0.5.7: Add performInitialScan() Function +### Step 0.5.8: Add performInitialScan() Function **File**: `internal/services/media_scanner.go` **Action**: Add after scanDirectory(): @@ -665,22 +731,38 @@ func (s *MediaScanner) performInitialScan(ctx context.Context) { continue } - // Scan directory - s.scanDirectory(ctx, folder) + // Submit scan job to worker (non-blocking) + job := &Job{ + ID: uuid.New().String(), + Type: JobTypeDirectoryScan, + Params: map[string]interface{}{ + "directory": folder, + "db": s.db, + }, + Status: JobStatusPending, + } + + if WorkerInstance != nil { + WorkerInstance.Enqueue(job) + fmt.Printf("Enqueued initial scan job: %s\n", folder) + } else { + fmt.Printf("Warning: Worker not initialized, skipping initial scan: %s\n", folder) + } } - fmt.Printf("Initial scan complete. Found %d new items\n", s.newItems) + fmt.Printf("Initial scan jobs enqueued\n") } ``` **Why**: Ensures existing files are detected when watching starts, not just new changes. -Without this, the system would miss all existing files until the 60-second polling runs. +Submits jobs to worker instead of calling scanDirectory() directly, ensuring all scans +go through the job queue for proper concurrency control. **Verification**: `go build ./internal/services/` --- -### Step 0.5.8: Add Close() Method for Cleanup +### Step 0.5.9: Add Close() Method for Cleanup **File**: `internal/services/media_scanner.go` **Action**: Add after performInitialScan(): @@ -694,9 +776,9 @@ func (s *MediaScanner) Close() error { s.watcher.Close() } - // FIXED: Clean up fileStability map to prevent memory leaks + // Clean up fileStability map to prevent memory leaks s.fileStabilityMu.Lock() - s.fileStability = make(map[string]time.Time) // Clear all entries + s.fileStability = make(map[string]*atomic.Bool) s.fileStabilityMu.Unlock() // Clear dirty directories @@ -704,26 +786,21 @@ func (s *MediaScanner) Close() error { s.dirtyDirs = make(map[string]time.Time) s.dirtyDirsMu.Unlock() - // Wait for active scans to complete (with timeout) + // Wait for in-progress scan to complete (with timeout) timeout := time.After(5 * time.Second) done := make(chan struct{}) go func() { - s.activeScansMu.Lock() - for len(s.activeScans) > 0 { - s.activeScansMu.Unlock() - time.Sleep(100 * time.Millisecond) - s.activeScansMu.Lock() - } - s.activeScansMu.Unlock() + s.scan_mutex.Lock() + s.scan_mutex.Unlock() close(done) }() select { case <-done: - fmt.Printf("All active scans completed\n") + fmt.Printf("Scanner cleanup complete\n") case <-timeout: - fmt.Printf("Timeout waiting for active scans\n") + fmt.Printf("Timeout waiting for scan to complete\n") } return nil @@ -733,13 +810,13 @@ func (s *MediaScanner) Close() error { **Why**: Ensures clean shutdown without resource leaks. Cleans up: - fileStability map (prevents memory leaks) - dirtyDirs map -- Waits for active scans to complete (graceful shutdown) +- Waits for in-progress scan to complete (simpler than activeScans map) **Verification**: `go build ./internal/services/` --- -### Step 0.5.9: Add Unit Tests +### Step 0.5.10: Add Unit Tests **File**: `internal/services/media_scanner_test.go` (new) **Action**: Create comprehensive unit tests: @@ -895,7 +972,7 @@ func TestProcessDirtyDirectories_BatchesScans(t *testing.T) { --- -### Step 0.5.9: Add Test Isolation with Snapshot/Restore +### Step 0.5.11: Add Test Isolation with Snapshot/Restore **File**: `cmd/server/tests/test_helpers.go` **Location**: In `setupTestServer()` function @@ -942,7 +1019,7 @@ t.Cleanup(func() { --- -### Step 0.5.9.5: Fix Test Expectation (60 → 300) +### Step 0.5.12: Fix Test Expectation (60 → 300) **File**: `cmd/server/tests/scan_settings_integration_test.go` **Location**: Line 34 @@ -967,7 +1044,7 @@ podman compose --profile tests run --rm tests go test -v -run "TestScanSettings_ --- -### Step 0.5.10: Add Integration Tests +### Step 0.5.13: Add Integration Tests **File**: `cmd/server/tests/fsnotify_integration_test.go` (new) **Action**: Create integration tests using test_helpers: @@ -1069,7 +1146,7 @@ func TestFSNotify_BulkFileDetection(t *testing.T) { --- -### Step 0.5.10: Verify Phase 0.5 +### Step 0.5.14: Verify Phase 0.5 **Action**: Run full test suite: ```bash