pkg/watch: deadlock test

pkg/watch/mux.go

@@ -115,10 +115,13 @@ func (m *Mux) loop() {
 	m.closeAll()
 }
 
+var testHookMuxDistribute = func() {}
+
 // distribute sends event to all watchers. Blocking.
 func (m *Mux) distribute(event Event) {
 	m.lock.Lock()
 	defer m.lock.Unlock()
+	testHookMuxDistribute()
 	for _, w := range m.watchers {
 		select {
 		case w.result <- event:

pkg/watch/mux_test.go

@@ -20,6 +20,7 @@ import (
 	"reflect"
 	"sync"
 	"testing"
+	"time"
 )
 
 type myType struct {
@@ -91,3 +92,24 @@ func TestMuxWatcherClose(t *testing.T) {
 	w.Stop()
 	w2.Stop()
 }
+
+func TestMuxWatcherStopDeadlock(t *testing.T) {
+	defer func(fn func()) { testHookMuxDistribute = fn }(testHookMuxDistribute)
+	sig, done := make(chan bool), make(chan bool)
+	testHookMuxDistribute = func() { sig <- true }
+	m := NewMux(0)
+	go func(w Interface) {
+		// Imagine this goroutine was receiving from w.ResultChan()
+		// until it received some signal and stopped watching.
+		<-sig
+		w.Stop()
+		close(done)
+	}(m.Watch())
+	m.Action(Added, &myType{})
+	select {
+	case <-time.After(5 * time.Second):
+		t.Error("timeout: deadlocked")
+	case <-done:
+	}
+	m.Shutdown()
+}