[mtls/fsnotify] fix some crashes, improve clarity around locks

Concurrency safety was never fsnotify's strong point.

Use RWMutex to guard watchHandlers.

diff --git a/mtls/fsnotify/fsnotify.go b/mtls/fsnotify/fsnotify.go
index 23f8fefe1397363b4018da6b74e3b766d03687c8..787d30ba1520ff7be085e32cf12ea5930c840818 100644 (file)
--- a/mtls/fsnotify/fsnotify.go
+++ b/mtls/fsnotify/fsnotify.go
@@ -62,9 +62,13 @@ var (
 )
 
 var (
+       // startWatcherMu guards the `watched` map.
        startWatcherMu sync.Mutex
-       gWatcher       *fsnotify.Watcher
-       logger         log.Logger
+
+       // handlerMu guards watchHandlers
+       handlerMu sync.RWMutex
+       gWatcher  *fsnotify.Watcher
+       logger    log.Logger
 )
 
 // Map of original path => notifyRule.
@@ -82,7 +86,10 @@ var pendingWrites *hashset.HashSet[string]
 
 // map of {watched symlink path => rule name}, used to keep track of which symlinks we added
 // watchers on in order to detect changes to watched files
-var symlinkPropagationMap map[string]*hashset.HashSet[string]
+var (
+       symlinkPropagationMap map[string]*hashset.HashSet[string]
+       symlinkMu             sync.RWMutex
+)
 
 func startWatcher() error {
        startWatcherMu.Lock()
@@ -178,9 +185,9 @@ func Unsubscribe(filePath string) error {
 }
 
 func watcherLoop() {
-       defer (func() {
+       defer func() {
                gWatcher = nil
-       })()
+       }()
 
        for {
                select {
@@ -222,11 +229,19 @@ func handleEvent(event fsnotify.Event) {
                delete(watched, event.Name)
                startWatcherMu.Unlock()
        }
+
+       handlerMu.RLock()
        if _, ok := watchHandlers[event.Name]; ok {
                // Event occurred on the original path.
+               handlerMu.RUnlock()
                handleEventForPath(event, event.Name)
                return
-       } else if children, ok := symlinkPropagationMap[event.Name]; ok {
+       }
+       handlerMu.RUnlock()
+
+       symlinkMu.RLock()
+       if children, ok := symlinkPropagationMap[event.Name]; ok {
+               symlinkMu.RUnlock()
                // Event occurred on the underlying file (recursively-resolved absolute path) of a watched
                // path
                for _, child := range children.AsSortedSlice() {
@@ -235,7 +250,10 @@ func handleEvent(event fsnotify.Event) {
                        handleEventForPath(event, child)
                }
                return
-       } else if links := symlinkPropagationChildren(event.Name); len(links) > 0 {
+       }
+       symlinkMu.RUnlock()
+
+       if links := symlinkPropagationChildren(event.Name); len(links) > 0 {
                // Event occurred on a symlink in the directory chain of the watched path or a link target
                // between the original file and the underlying absolute path. Practical example:
                //
@@ -263,12 +281,18 @@ func handleEvent(event fsnotify.Event) {
                // In these cases we pick up the event on `/config/..data` and fire events for any original
                // paths
                for _, link := range links {
-                       for _, ruleName := range symlinkPropagationMap[link].AsSlice() {
+                       symlinkMu.RLock()
+                       children := symlinkPropagationMap[link]
+                       symlinkMu.RUnlock()
+                       for _, ruleName := range children.AsSlice() {
+                               handlerMu.RLock()
                                if _, ok := watchHandlers[ruleName]; ok {
+                                       handlerMu.RUnlock()
                                        logger.Debugf("propagating %v event on %s to origin %s",
                                                event.Op, link, ruleName)
                                        handleEventForPath(event, ruleName)
                                } else {
+                                       handlerMu.RUnlock()
                                        logger.Errorf("symlinkPropagationMap target without corresponding rule: %s", ruleName)
                                }
                        }
@@ -278,6 +302,8 @@ func handleEvent(event fsnotify.Event) {
        }
        // Don't issue this warning if we received an event for something being created in a directory
        // we're watching because of files under it
+       startWatcherMu.Lock()
+       defer startWatcherMu.Unlock()
        if _, ok := watched[path.Dir(event.Name)]; ok {
                if event.Has(fsnotify.Create) || event.Has(fsnotify.Remove) || event.Has(fsnotify.Rename) {
                        return
@@ -293,7 +319,9 @@ func handleEvent(event fsnotify.Event) {
 // multiple watched files is a symlink that gets repointed. For these cases, handleEventForPath is
 // called multiple times from a single invocation of handleEvent.
 func handleEventForPath(event fsnotify.Event, ruleName string) {
+       handlerMu.RLock()
        rule, ok := watchHandlers[ruleName]
+       handlerMu.RUnlock()
        if !ok {
                logger.Errorf("handleEventForPath called for file without rule: %s: %+v", ruleName, event)
                return
@@ -363,6 +391,9 @@ func trackSymlink(realPath, ruleName string) {
        if realPath == ruleName {
                return
        }
+       symlinkMu.Lock()
+       defer symlinkMu.Unlock()
+
        if _, ok := symlinkPropagationMap[realPath]; !ok {
                symlinkPropagationMap[realPath] = hashset.NewHashSet[string]()
        }
@@ -399,7 +430,9 @@ func symlinkPropagationChildren(link string) []string {
 }
 
 func scanSymlinks(filePath string) error {
+       handlerMu.RLock()
        rule := watchHandlers[filePath]
+       handlerMu.RUnlock()
        rule.upstreams.Empty()
        target := filePath
        for {
@@ -452,12 +485,14 @@ func Debug() {
        }
 
        logger.Debugf("watchHandlers:")
+       handlerMu.RLock()
        if len(watchHandlers) == 0 {
                logger.Debugf("  <NONE>")
        }
        for filePath, rule := range watchHandlers {
                logger.Debugf("   %s: upstreams=%+v, %+v", filePath, rule.upstreams.AsSortedSlice(), rule.callbacks)
        }
+       handlerMu.RUnlock()
 
        logger.Debugf("fsnotify ref counts:")
        if len(watched) == 0 {
@@ -485,14 +520,15 @@ func NotifyPath(filePath string, callback NotifyFunc) error {
                return err
        }
 
+       handlerMu.Lock()
        if _, ok := watchHandlers[filePath]; !ok {
                watchHandlers[filePath] = &notifyRule{
                        upstreams: hashset.NewHashSet[string](),
                }
        }
        rule := watchHandlers[filePath]
-
        rule.callbacks = append(watchHandlers[filePath].callbacks, callback)
+       handlerMu.Unlock()
 
        scanSymlinks(filePath)