Simplify file provider synchronization to use only mutex

- Remove atomic operations in favor of simple boolean flags
- Fix potential deadlock in Watch() by releasing lock before goroutine spawn
- Use minimal locking in cleanup goroutine
- Cleaner, more maintainable synchronization pattern
- All tests pass with race detector

Author: rhnvrm

providers/file/file.go

@@ -9,7 +9,6 @@ import (
 	"os"
 	"path/filepath"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/fsnotify/fsnotify"
@@ -21,11 +20,8 @@ type File struct {
 	w    *fsnotify.Watcher
 
 	// Mutex to protect concurrent access to watcher state
-	mu sync.Mutex
-
-	// Using Go 1.18 atomic functions for backwards compatibility.
-	isWatching  uint32
-	isUnwatched uint32
+	mu         sync.Mutex
+	isWatching bool
 }
 
 // Provider returns a file provider.
@@ -47,10 +43,10 @@ func (f *File) Read() (map[string]interface{}, error) {
 // blocking function that internally spawns a goroutine to watch for changes.
 func (f *File) Watch(cb func(event interface{}, err error)) error {
 	f.mu.Lock()
-	defer f.mu.Unlock()
-
+	
 	// If a watcher already exists, return an error.
-	if atomic.LoadUint32(&f.isWatching) == 1 {
+	if f.isWatching {
+		f.mu.Unlock()
 		return errors.New("file is already being watched")
 	}
 
@@ -68,10 +64,24 @@ func (f *File) Watch(cb func(event interface{}, err error)) error {
 
 	f.w, err = fsnotify.NewWatcher()
 	if err != nil {
+		f.mu.Unlock()
 		return err
 	}
 
-	atomic.StoreUint32(&f.isWatching, 1)
+	f.isWatching = true
+	
+	// Set up the directory watch before releasing the lock
+	err = f.w.Add(fDir)
+	if err != nil {
+		f.w.Close()
+		f.w = nil
+		f.isWatching = false
+		f.mu.Unlock()
+		return err
+	}
+	
+	// Release the lock before spawning goroutine
+	f.mu.Unlock()
 
 	var (
 		lastEvent     string
@@ -84,8 +94,12 @@ func (f *File) Watch(cb func(event interface{}, err error)) error {
 			select {
 			case event, ok := <-f.w.Events:
 				if !ok {
-					// Only throw an error if it was not an explicit unwatch.
-					if atomic.LoadUint32(&f.isUnwatched) == 0 {
+					// Only throw an error if we were still supposed to be watching.
+					f.mu.Lock()
+					stillWatching := f.isWatching
+					f.mu.Unlock()
+					
+					if stillWatching {
 						cb(nil, errors.New("fsnotify watch channel closed"))
 					}
 
@@ -133,8 +147,12 @@ func (f *File) Watch(cb func(event interface{}, err error)) error {
 			// There's an error.
 			case err, ok := <-f.w.Errors:
 				if !ok {
-					// Only throw an error if it was not an explicit unwatch.
-					if atomic.LoadUint32(&f.isUnwatched) == 0 {
+					// Only throw an error if we were still supposed to be watching.
+					f.mu.Lock()
+					stillWatching := f.isWatching
+					f.mu.Unlock()
+					
+					if stillWatching {
 						cb(nil, errors.New("fsnotify err channel closed"))
 					}
 
@@ -148,27 +166,32 @@ func (f *File) Watch(cb func(event interface{}, err error)) error {
 		}
 
 		f.mu.Lock()
-		atomic.StoreUint32(&f.isWatching, 0)
-		atomic.StoreUint32(&f.isUnwatched, 0)
+		f.isWatching = false
 		if f.w != nil {
 			f.w.Close()
 			f.w = nil
 		}
 		f.mu.Unlock()
 	}()
 
-	// Watch the directory for changes.
-	return f.w.Add(fDir)
+	return nil
 }
 
 // Unwatch stops watching the files and closes fsnotify watcher.
 func (f *File) Unwatch() error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 
-	atomic.StoreUint32(&f.isUnwatched, 1)
+	if !f.isWatching {
+		return nil // Already unwatched
+	}
+	
+	f.isWatching = false
 	if f.w != nil {
+		// Close the watcher to signal the goroutine to stop
+		// The goroutine will handle setting f.w = nil
 		return f.w.Close()
 	}
-	return nil
+	// This state should ideally never be reached - it indicates a bug in the synchronization logic
+	return errors.New("file watcher is in an inconsistent state: isWatching is true but watcher is nil")
 }

tests/koanf_test.go

@@ -1942,3 +1942,104 @@ func TestNoDeadlock(t *testing.T) {
 		t.Fatal("DEADLOCK DETECTED: Goroutines did not complete within timeout")
 	}
 }
+
+// TestFileProviderConcurrency specifically tests the file provider's synchronization
+// under heavy concurrent load to catch any races in Watch/Unwatch
+func TestFileProviderConcurrency(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping file provider concurrency test in short mode")
+	}
+
+	// Create temp config file
+	tmpDir := t.TempDir()
+	tmpFile := filepath.Join(tmpDir, "config.json")
+	initialData := []byte(`{"test": "value"}`)
+	err := os.WriteFile(tmpFile, initialData, 0600)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var wg sync.WaitGroup
+	done := make(chan struct{})
+
+	// Stress test: multiple goroutines rapidly watch/unwatch the same file
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+			
+			for {
+				select {
+				case <-done:
+					return
+				default:
+					// Create new file provider for each iteration
+					f := file.Provider(tmpFile)
+					
+					// Try to watch
+					var watchErr error
+					watchErr = f.Watch(func(event interface{}, err error) {
+						// Simple callback that doesn't do much
+						_ = event
+						_ = err
+					})
+					
+					// Sometimes watch will fail if another goroutine is already watching
+					if watchErr == nil {
+						// If watch succeeded, unwatch after a short delay
+						time.Sleep(time.Millisecond)
+						unwatchErr := f.Unwatch()
+						if unwatchErr != nil {
+							// Log but don't fail - this can happen during cleanup
+							t.Logf("Unwatch error (normal during stress test): %v", unwatchErr)
+						}
+					} else if watchErr.Error() != "file is already being watched" {
+						// Unexpected error
+						t.Errorf("Unexpected watch error: %v", watchErr)
+					}
+					
+					// Small delay to prevent tight loop
+					time.Sleep(time.Microsecond)
+				}
+			}
+		}(i)
+	}
+
+	// Also test concurrent file modifications while watching
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		counter := 0
+		
+		for {
+			select {
+			case <-done:
+				return
+			default:
+				// Write new data to trigger file events
+				newData := fmt.Sprintf(`{"test": "value", "counter": %d}`, counter)
+				os.WriteFile(tmpFile, []byte(newData), 0600)
+				counter++
+				time.Sleep(5 * time.Millisecond)
+			}
+		}
+	}()
+
+	// Let the stress test run for 200ms
+	time.Sleep(200 * time.Millisecond)
+	close(done)
+
+	// Wait with timeout to detect deadlocks
+	waitChan := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(waitChan)
+	}()
+
+	select {
+	case <-waitChan:
+		t.Log("File provider concurrency test completed successfully")
+	case <-time.After(5 * time.Second):
+		t.Fatal("FILE PROVIDER DEADLOCK: Goroutines did not complete within timeout")
+	}
+}