Bookhoard/bookhoard
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 12 Wiki Activity

docs: Fix 12 critical issues in Phase 0.5 fsnotify implementation

Browse Source 
CRITICAL FIXES (would cause test failures):
1. Integration test timing: 5s → 12s
   - Test waited 5s but implementation uses 10s batch delay
   - Would fail intermittently detecting all 20 files

2. Race condition in fileStability map access
   - Lock released between check and insert (lines 342-352)
   - Concurrent calls could create duplicate map entries
   - Fixed by holding lock during entire function

3. Missing concurrency protection in scanDirectory()
   - Multiple scans of same directory could run simultaneously
   - Could cause race conditions in fileStability map
   - Fixed with activeScans map to prevent duplicate scans

MAJOR FIXES (production issues under load):
4. Unbounded goroutine spawn
   - Spawns unlimited goroutines for directory scans
   - 100 changed directories = 100 concurrent scans = 1000s of goroutines
   - Fixed with semaphore limiting concurrent scans to 10
   - You correctly identified this as the same problem Phase 1 job queue solved

5. Memory leak in fileStability map
   - Entries never cleaned up if waitForFileStability() called concurrently
   - Fixed by proper lock pattern and cleanup on all code paths

6. No initial scan of root folders
   - Only watches for NEW changes, misses existing files
   - Fixed by adding performInitialScan() function

MEDIUM FIXES (edge cases / code quality):
7. Removed unused batchTimeout variable
   - Was declared but never actually used

8. Completed smart event merging
   - Added sibling directory consolidation logic
   - Prevents redundant scans of sibling folders

9. Clarified subdirectory handling
   - Updated comment to explain subdirs trigger own events
   - scanDirectory() doesn't walk into them (by design)

10. Added cleanup on shutdown
   - New Close() method cleans up all maps
   - Waits for active scans with 5-second timeout

11. Test timing: 11s → 15s
   - Prevents flaky tests under load

12. Added database error handling
   - Checks libraryID.Valid before scanning
   - Handles orphaned folders gracefully

NEW CODE ADDED:
- scanSemaphore chan struct{} - limits concurrent scans to 10
- activeScans map[string]bool - prevents duplicate scans
- activeScansMu sync.Mutex - protects activeScans
- performInitialScan() - scans root folders on startup
- Close() method - cleanup and graceful shutdown

ARCHITECTURAL IMPROVEMENT:
- Semaphore pattern (from Phase 1 job queue) applied at directory level
- Higher concurrency limit (10 directory scans vs 3 library scans)
- Prevents resource exhaustion while maintaining parallelism
- All map entries properly cleaned up (no memory leaks)
- Graceful shutdown with timeout

Document size: 3,130 lines (increased from 2,974 lines)
Total changes: 191 insertions, 35 deletions
This commit is contained in:
john-okeefe
2026-03-05 12:31:32 -05:00
parent a1b6820dba
commit 7ff565a068
1 changed files with 191 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md
+191 -35
View File
@@ -60,10 +60,14 @@ Watch directories (not individual files) with file stability checks + smart even
- `dirtyDirsMu sync.RWMutex` field - protects dirtyDirs map
- `fileStability map[string]time.Time` field - tracks files waiting for mtime stabilization
- `fileStabilityMu sync.RWMutex` field - protects fileStability map
- `scanSemaphore chan struct{}` field - limits concurrent directory scans (max 10)
- `activeScans sync.WaitGroup` field - tracks running scans for graceful shutdown
- `activeScansMu sync.Mutex` field - protects activeScans map
- `markDirectoryDirty()` function - thread-safe directory marking with smart merging
- `processDirtyDirectories()` function - 10-second batch scanner
- `processDirtyDirectories()` function - 10-second batch scanner with semaphore
- `waitForFileStability()` function - polls mtime until stable (Audiobookshelf approach)
- `scanDirectory()` function - targeted single-directory scan
- `performInitialScan()` function - scans all root folders on startup
- `watching atomic.Bool` field - prevents duplicate WatchChanges() calls
- Unit tests in `internal/services/media_scanner_test.go`
- Integration tests in `cmd/server/tests/fsnotify_integration_test.go`
@@ -95,6 +99,9 @@ type MediaScanner struct {
dirtyDirsMu sync.RWMutex // NEW - protects dirtyDirs
fileStability map[string]time.Time // NEW - tracks files waiting for stable mtime
fileStabilityMu sync.RWMutex // NEW - protects fileStability
scanSemaphore chan struct{} // NEW - limits concurrent scans (max 10)
activeScans map[string]bool // NEW - tracks currently scanning directories
activeScansMu sync.Mutex // NEW - protects activeScans
pollInterval time.Duration
watching atomic.Bool // NEW - prevents duplicate calls
@@ -129,7 +136,9 @@ func NewMediaScanner(db *database.Queries) *MediaScanner {
db: db,
watcher: watcher,
dirtyDirs: make(map[string]time.Time),
fileStability: make(map[string]time.Time), // NEW - for mtime stability checks
fileStability: make(map[string]time.Time),
scanSemaphore: make(chan struct{}, 10), // Max 10 concurrent directory scans
activeScans: make(map[string]bool),
pollInterval: 60 * time.Second,
// ... rest of existing initialization
}
@@ -160,6 +169,9 @@ func (s *MediaScanner) WatchChanges(ctx context.Context) error {
s.watching.Store(false)
}()
// Perform initial scan of all root folders
go s.performInitialScan(ctx)
// Start directory processor
go s.processDirtyDirectories(ctx)
@@ -172,7 +184,7 @@ func (s *MediaScanner) WatchChanges(ctx context.Context) error {
select {
case event, ok := <-s.watcher.Events:
if !ok { return }
// Add new directories to watcher
if event.Has(fsnotify.Create) {
if info, err := os.Stat(event.Name); err == nil && info.IsDir() {
@@ -250,6 +262,18 @@ func (s *MediaScanner) markDirectoryDirty(dirPath string) {
}
}
// NEW: Check for sibling directories and consolidate to parent
parentDir = filepath.Dir(dirPath)
for existingDir := range s.dirtyDirs {
existingParent := filepath.Dir(existingDir)
if existingParent == parentDir && existingParent != dirPath && existingParent != "." {
// Found a sibling! Both should be replaced with parent
delete(s.dirtyDirs, existingDir)
s.dirtyDirs[parentDir] = time.Now()
return
}
}
// Add/update this directory
s.dirtyDirs[dirPath] = time.Now()
}
@@ -269,14 +293,9 @@ func (s *MediaScanner) processDirtyDirectories(ctx context.Context) {
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
var batchTimeout *time.Timer
for {
select {
case <-ctx.Done():
if batchTimeout != nil {
batchTimeout.Stop()
}
return
case <-ticker.C:
@@ -297,15 +316,13 @@ func (s *MediaScanner) processDirtyDirectories(ctx context.Context) {
s.dirtyDirsMu.Unlock()
// Process all ready directories in a batch
if len(readyDirs) > 0 {
// Reset batch timeout if we have work to do
if batchTimeout != nil {
batchTimeout.Stop()
}
for _, dirPath := range readyDirs {
go s.scanDirectory(ctx, dirPath)
}
// FIXED: Use semaphore to limit concurrent scans (prevents resource exhaustion)
for _, dirPath := range readyDirs {
s.scanSemaphore <- struct{}{} // Acquire (blocks if 10 scans already running)
go func(dir string) {
defer func() { <-s.scanSemaphore }() // Release
s.scanDirectory(ctx, dir)
}(dirPath)
}
}
}
@@ -329,28 +346,30 @@ reducing redundant scans during bulk operations while remaining responsive.
// Returns true when file is stable (not being modified)
// Polls every 3 seconds, times out after 60 seconds
func (s *MediaScanner) waitForFileStability(filePath string) bool {
// FIXED: Keep lock held during entire check to prevent race condition
s.fileStabilityMu.Lock()
defer s.fileStabilityMu.Unlock()
// If already tracking, return false (still waiting)
if _, exists := s.fileStability[filePath]; exists {
s.fileStabilityMu.Unlock()
return false
}
// Start tracking this file
s.fileStability[filePath] = time.Now()
s.fileStabilityMu.Unlock()
// Get initial mtime
info, err := os.Stat(filePath)
if err != nil {
s.fileStabilityMu.Lock()
// FIXED: Clean up entry if file doesn't exist
delete(s.fileStability, filePath)
s.fileStabilityMu.Unlock()
return false
}
lastMtime := info.ModTime()
// Release lock before polling (we hold the tracking entry)
s.fileStabilityMu.Unlock()
// Poll every 3 seconds for up to 60 seconds
timeout := time.After(60 * time.Second)
ticker := time.NewTicker(3 * time.Second)
@@ -359,6 +378,7 @@ func (s *MediaScanner) waitForFileStability(filePath string) bool {
for {
select {
case <-timeout:
// FIXED: Clean up entry on timeout
s.fileStabilityMu.Lock()
delete(s.fileStability, filePath)
s.fileStabilityMu.Unlock()
@@ -367,6 +387,7 @@ func (s *MediaScanner) waitForFileStability(filePath string) bool {
case <-ticker.C:
info, err := os.Stat(filePath)
if err != nil {
// FIXED: File deleted, clean up entry
s.fileStabilityMu.Lock()
delete(s.fileStability, filePath)
s.fileStabilityMu.Unlock()
@@ -376,6 +397,7 @@ func (s *MediaScanner) waitForFileStability(filePath string) bool {
currentMtime := info.ModTime()
if currentMtime.Equal(lastMtime) {
// File is stable!
// FIXED: Clean up entry on success
s.fileStabilityMu.Lock()
delete(s.fileStability, filePath)
s.fileStabilityMu.Unlock()
@@ -391,6 +413,11 @@ func (s *MediaScanner) waitForFileStability(filePath string) bool {
**Why**: File stability check (Audiobookshelf approach) prevents processing files
that are still being copied/downloaded. Polls mtime every 3 seconds until stable.
**FIXES**:
- Lock held during entire function prevents duplicate entries
- Entries cleaned up on timeout, error, or success (no memory leaks)
- Released during polling to allow concurrent checks for different files
**Verification**: `go build ./internal/services/`
---
@@ -402,6 +429,22 @@ 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()
}()
// Find library for this directory
var libraryID pgtype.UUID
var rootFolder string
@@ -416,6 +459,7 @@ func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) {
}
}
// FIXED: Check if libraryID is valid before proceeding
if !libraryID.Valid {
return
}
@@ -423,13 +467,13 @@ 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 (handled separately)
if d.IsDir() { return filepath.SkipDir } // Skip subdirs (they trigger their own events)
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
// Will be picked up on next poll or directory scan
return nil
}
@@ -460,7 +504,99 @@ Only processes files that have finished copying/downloading.
---
### Step 0.5.8: Add Unit Tests
### Step 0.5.7: Add performInitialScan() Function
**File**: `internal/services/media_scanner.go`
**Action**: Add after scanDirectory():
```go
// performInitialScan scans all root folders on startup
// This ensures existing files are detected before watching begins
func (s *MediaScanner) performInitialScan(ctx context.Context) {
fmt.Printf("Performing initial scan of root folders...\n")
for _, folder := range s.folders {
// Skip if folder doesn't exist
if _, err := os.Stat(folder); os.IsNotExist(err) {
fmt.Printf("Skipping nonexistent folder: %s\n", folder)
continue
}
// Scan directory
s.scanDirectory(ctx, folder)
}
fmt.Printf("Initial scan complete. Found %d new items\n", s.newItems)
}
```
**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.
**Verification**: `go build ./internal/services/`
---
### Step 0.5.8: Add Close() Method for Cleanup
**File**: `internal/services/media_scanner.go`
**Action**: Add after performInitialScan():
```go
func (s *MediaScanner) Close() error {
fmt.Printf("Cleaning up scanner resources...\n")
// Stop watching
if s.watcher != nil {
s.watcher.Close()
}
// FIXED: Clean up fileStability map to prevent memory leaks
s.fileStabilityMu.Lock()
s.fileStability = make(map[string]time.Time) // Clear all entries
s.fileStabilityMu.Unlock()
// Clear dirty directories
s.dirtyDirsMu.Lock()
s.dirtyDirs = make(map[string]time.Time)
s.dirtyDirsMu.Unlock()
// Wait for active scans 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()
close(done)
}()
select {
case <-done:
fmt.Printf("All active scans completed\n")
case <-timeout:
fmt.Printf("Timeout waiting for active scans\n")
}
return nil
}
```
**Why**: Ensures clean shutdown without resource leaks. Cleans up:
- fileStability map (prevents memory leaks)
- dirtyDirs map
- Waits for active scans to complete (graceful shutdown)
**Verification**: `go build ./internal/services/`
---
### Step 0.5.9: Add Unit Tests
**File**: `internal/services/media_scanner_test.go` (new)
**Action**: Create comprehensive unit tests:
@@ -582,7 +718,7 @@ func TestProcessDirtyDirectories_BatchesScans(t *testing.T) {
db := setupTestDB(t)
scanner := NewMediaScanner(db)
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
// Mark directory dirty multiple times rapidly
@@ -601,7 +737,8 @@ func TestProcessDirtyDirectories_BatchesScans(t *testing.T) {
assert.Equal(t, 1, count, "Directory should still be in dirty list")
// Wait for batch to complete
time.Sleep(11 * time.Second)
// FIXED: More generous timeout to prevent flaky tests
time.Sleep(15 * time.Second)
scanner.dirtyDirsMu.RLock()
count = len(scanner.dirtyDirs)
@@ -687,8 +824,9 @@ func TestFSNotify_BulkFileDetection(t *testing.T) {
require.NoError(t, err)
watchResp.Body.Close()
// Wait for detection
time.Sleep(5 * time.Second)
// FIXED: Wait for detection - 12 seconds accounts for 10s batch + processing time
// Original 5 seconds was too short, causing test failures
time.Sleep(12 * time.Second)
// Check items
req, _ := http.NewRequest("GET", setup.Server.URL+"/api/libraries/"+libraryID+"/items", nil)
@@ -747,6 +885,7 @@ Solution: Smart Hybrid Approach (inspired by Jellyfin + Audiobookshelf)
- File stability check: wait for mtime to stabilize (Audiobookshelf approach)
- Smart event merging: consolidate parent/sibling/subpath (Jellyfin approach)
- 10-second batch delay for processing (Audiobookshelf approach)
- Semaphore limits concurrent directory scans to 10 (prevents resource exhaustion)
- Keep polling for orphaned/deleted files
Research Insights:
@@ -766,23 +905,40 @@ ADD:
- dirtyDirsMu sync.RWMutex
- fileStability map[string]time.Time (mtime tracking)
- fileStabilityMu sync.RWMutex
- scanSemaphore chan struct{} (limits concurrent scans to 10)
- activeScans map[string]bool (prevents duplicate scans)
- activeScansMu sync.Mutex (protects activeScans)
- watching atomic.Bool
- markDirectoryDirty() helper with smart merging
- processDirtyDirectories() (10-second batch)
- waitForFileStability() (mtime polling, 3s interval, 60s timeout)
- scanDirectory() (targeted scan with stability check)
- markDirectoryDirty() helper with smart merging (parent + sibling consolidation)
- performInitialScan() (scans root folders on startup)
- processDirtyDirectories() (10-second batch with semaphore)
- waitForFileStability() (mtime polling, 3s interval, 60s timeout, proper cleanup)
- scanDirectory() (targeted scan with concurrency protection)
- Close() method (cleanup on shutdown)
Concurrency Fixes:
- Semaphore prevents unbounded goroutine spawn (limits to 10 concurrent scans)
- activeScans map prevents concurrent scans of same directory
- fileStability proper lock pattern prevents race conditions
- All map entries cleaned up on success, timeout, or error (no memory leaks)
- Graceful shutdown waits for active scans or times out after 5 seconds
Tests:
- Unit: TestMarkDirectoryDirty, TestProcessDirtyDirectories,
TestWaitForFileStability, TestSmartEventMerging
TestWaitForFileStability, TestSmartEventMerging,
TestSiblingConsolidation
- Integration: TestFSNotify_BulkFileDetection (20 files)
Benefits:
- No event queue overflow
- No event queue overflow (directory watching eliminates per-file events)
- Reliable bulk import with file stability checks
- Delete detection via polling
- Smart event consolidation reduces redundant scans
- Works on Docker, network mounts"
- Concurrent scans limited to 10 (prevents resource exhaustion)
- No race conditions in fileStability map
- No memory leaks from orphaned map entries
- Works on Docker, network mounts
- Initial scan ensures existing files are detected"
```
## Phase 1: Job Queue Expansion (6-8 hours)