fix: prevent oathkeeper rule fetcher watch deadlock

GitOrigin-RevId: 179003cbf9bffe7e234761a0a20724738cd05f2a

Author: aeneasr

oryx/watcherx/definitions.go

@@ -7,6 +7,8 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+
+	"github.com/pkg/errors"
 )
 
 type (
@@ -22,6 +24,7 @@ type (
 		DispatchNow() (<-chan int, error)
 	}
 	dispatcher struct {
+		ctx     context.Context
 		trigger chan struct{}
 		done    chan int
 	}
@@ -42,8 +45,9 @@ func (e *errSchemeUnknown) Error() string {
 	return fmt.Sprintf("unknown scheme '%s' to watch", e.scheme)
 }
 
-func newDispatcher() *dispatcher {
+func newDispatcher(ctx context.Context) *dispatcher {
 	return &dispatcher{
+		ctx:     ctx,
 		trigger: make(chan struct{}),
 		done:    make(chan int),
 	}
@@ -53,8 +57,15 @@ func (d *dispatcher) DispatchNow() (<-chan int, error) {
 	if d.trigger == nil {
 		return nil, ErrWatcherNotRunning
 	}
-	d.trigger <- struct{}{}
-	return d.done, nil
+	// The trigger send must respect cancellation. Once the watcher's context is
+	// done its receiver goroutine has returned, so a bare send would block
+	// forever and wedge the caller.
+	select {
+	case d.trigger <- struct{}{}:
+		return d.done, nil
+	case <-d.ctx.Done():
+		return nil, errors.WithStack(ErrWatcherNotRunning)
+	}
 }
 
 func Watch(ctx context.Context, u *url.URL, c EventChannel) (Watcher, error) {

oryx/watcherx/directory.go

@@ -35,7 +35,7 @@ func WatchDirectory(ctx context.Context, dir string, c EventChannel) (Watcher, e
 	}
 
 	dw := &directoryWatcher{
-		dispatcher: newDispatcher(),
+		dispatcher: newDispatcher(ctx),
 		c:          c,
 		dir:        dir,
 		subDirs:    subDirs,

oryx/watcherx/file.go

@@ -39,7 +39,7 @@ func WatchFile(ctx context.Context, file string, c EventChannel) (Watcher, error
 			return nil, errors.WithStack(err)
 		}
 	}
-	d := newDispatcher()
+	d := newDispatcher(ctx)
 	go streamFileEvents(ctx, watcher, c, d.trigger, d.done, file, resolvedFile)
 	return d, nil
 }

rule/fetcher_default.go

@@ -102,8 +102,12 @@ func (f *FetcherDefault) watchLocalFiles(ctx context.Context) {
 		if cancel, ok := f.cancelWatchers[fp]; !ok {
 			// watch all files we are not yet watching
 			repoChanged = true
-			ctx, cancelWatchers[fp] = context.WithCancel(ctx)
-			w, err := watcherx.WatchFile(ctx, fp, f.events)
+			// Derive each file's context from the caller's ctx, not from the
+			// previous iteration's, so cancelling one file's watcher does not
+			// cancel the others.
+			var fileCtx context.Context
+			fileCtx, cancelWatchers[fp] = context.WithCancel(ctx)
+			w, err := watcherx.WatchFile(fileCtx, fp, f.events)
 			if err != nil {
 				f.registry.Logger().WithError(err).WithField("file", fp).Error("Unable to watch file, ignoring it.")
 				continue
@@ -144,6 +148,12 @@ func (f *FetcherDefault) watchLocalFiles(ctx context.Context) {
 }
 
 func (f *FetcherDefault) Watch(ctx context.Context) error {
+	// Start the local update consumer before watching files. watchLocalFiles
+	// forces an initial read via DispatchNow, which sends on the unbuffered
+	// f.events channel; without a reader already running, that send can block
+	// and deadlock the watcher setup.
+	go f.processLocalUpdates(ctx)
+
 	f.watchLocalFiles(ctx)
 
 	getRemoteRepos := func() map[url.URL]struct{} {
@@ -191,7 +201,6 @@ func (f *FetcherDefault) Watch(ctx context.Context) error {
 		remoteRepos = newRemoteRepos
 	})
 
-	go f.processLocalUpdates(ctx)
 	return nil
 }