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

docs: Rewrite Phase 0.5 with clean architecture (Phase 0.5 + Phase 1 robustness)

Browse Source 
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
This commit is contained in:
john-okeefe
2026-03-05 12:55:48 -05:00
parent eecfb52996
commit 1a552e03f1
1 changed files with 151 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md
+151 -74
View File
@@ -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