Merge pull request #1716 from MaxRink/fix/fix-debouncer-deadlock

fix(debouncer): prevent deadlock when forwarding to a terminating object

pkg/debouncer/debouncer.go

@@ -52,9 +52,8 @@ func (n *ns) add(name string, output chan<- watch.Event) *object {
 	return i
 }
 
-func (n *ns) del(name string) {
-	if _, exists := n.Load(name); exists {
-		n.Delete(name)
+func (n *ns) del(name string, object *object) {
+	if n.CompareAndDelete(name, object) {
 		n.cnt.Add(-1)
 	}
 }
@@ -119,35 +118,57 @@ func (d *debouncer) Start(ctx context.Context) error {
 				return fmt.Errorf("objects of type %T are not supported", v)
 			}
 
-			eventNs, exists := d.getNs(namespace)
-			if !exists {
-				// if not, create new map for the namespace
-				eventNs = d.addNs(namespace)
-			}
+		processEvent:
+			for {
+				eventNs, exists := d.getNs(namespace)
+				if !exists {
+					// if not, create new map for the namespace
+					eventNs = d.addNs(namespace)
+				}
 
-			// check if the object was previously reconciled
-			eventObject, exists := eventNs.get(name)
+				// check if the object was previously reconciled
+				eventObject, exists := eventNs.get(name)
 
-			// if not and the event is not of type 'Deleted', create new object
-			if !exists && tmp.Type != watch.Deleted {
-				eventObject = eventNs.add(name, d.output)
+				// if not and the event is not of type 'Deleted', create new object
+				if !exists && tmp.Type != watch.Deleted {
+					eventObject = eventNs.add(name, d.output)
 
-				wg.Go(func() {
-					// start deboucing events for this object
-					eventObject.start(debouncerCtx, d.debounceTime)
-					// if debouncer for the object ended - e.g. object was deleted - clean the map of objects
-					eventObject = nil
-					eventNs.del(name)
-					// if namespace is empty, delete the namespace map
-					if eventNs.cnt.Load() == 0 {
-						d.delNs(namespace)
+					workerObject := eventObject
+					workerNs := eventNs
+					workerName := name
+					workerNamespace := namespace
+					workerObject.onStop = func() {
+						// Remove the object before its worker can become receiver-less.
+						workerNs.del(workerName, workerObject)
 					}
-				})
-			}
 
-			if eventObject != nil {
+					wg.Go(func() {
+						// start deboucing events for this object
+						workerObject.start(debouncerCtx, d.debounceTime)
+						// if debouncer for the object ended - e.g. object was deleted - clean the map of objects
+						workerNs.del(workerName, workerObject)
+						// if namespace is empty, delete the namespace map
+						if workerNs.cnt.Load() == 0 {
+							d.delNs(workerNamespace, workerNs)
+						}
+					})
+				}
+
+				if eventObject == nil {
+					break processEvent
+				}
+
 				// pass the watch event to the debouncer object
-				eventObject.input <- tmp
+				select {
+				case eventObject.input <- tmp:
+					break processEvent
+				case <-eventObject.stopChan:
+					// The object stopped after the map lookup. Retry the event
+					// against the newly-created object instead of dropping it.
+					continue processEvent
+				case <-debouncerCtx.Done():
+					return nil
+				}
 			}
 		}
 	}
@@ -181,8 +202,8 @@ func (d *debouncer) addNs(namespace string) *ns {
 	return &n
 }
 
-func (d *debouncer) delNs(namespace string) {
-	d.namespaces.Delete(namespace)
+func (d *debouncer) delNs(namespace string, ns *ns) {
+	d.namespaces.CompareAndDelete(namespace, ns)
 }
 
 type object struct {
@@ -190,6 +211,7 @@ type object struct {
 	output   chan<- watch.Event
 	stopChan chan any
 	stopOnce sync.Once
+	onStop   func()
 }
 
 func newObject(output chan<- watch.Event) *object {
@@ -247,6 +269,9 @@ func (o *object) start(ctx context.Context, debounceTime time.Duration) {
 
 func (o *object) stop() {
 	o.stopOnce.Do(func() {
+		if o.onStop != nil {
+			o.onStop()
+		}
 		close(o.stopChan)
 	})
 }

pkg/debouncer/debouncer_deadlock_test.go

@@ -0,0 +1,169 @@
+package debouncer
+
+import (
+	"context"
+	"runtime"
+	"testing"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+)
+
+func TestStartReturnsWhenCancellationInterruptsObjectForwarding(t *testing.T) {
+	input := make(chan watch.Event)
+	d := &debouncer{
+		input:        input,
+		output:       make(chan watch.Event),
+		stopChan:     make(chan any),
+		debounceTime: 200 * time.Millisecond,
+	}
+
+	// Leave the object without a receiver. This is the state reached when its
+	// worker exits on context cancellation just before Start forwards an event.
+	ns := d.addNs("default")
+	object := ns.add("example", d.output)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- d.Start(ctx) }()
+
+	event := watch.Event{
+		Type: watch.Modified,
+		Object: &v1.Service{ObjectMeta: metav1.ObjectMeta{
+			Name: "example", Namespace: "default",
+		}},
+	}
+	sent := make(chan struct{})
+	go func() {
+		input <- event
+		close(sent)
+	}()
+	select {
+	case <-sent:
+	case <-time.After(250 * time.Millisecond):
+		cancel()
+		t.Fatal("debouncer did not receive the test event")
+	}
+	cancel()
+
+	select {
+	case <-done:
+		return
+	case <-object.input:
+		// Receiving here unblocks the production send. If cancellation had
+		// interrupted that send, Start would already have returned above.
+		select {
+		case <-done:
+		case <-time.After(250 * time.Millisecond):
+			t.Fatal("debouncer remained blocked forwarding an event after context cancellation")
+		}
+		t.Fatal("debouncer forwarded an event after context cancellation")
+	case <-time.After(250 * time.Millisecond):
+		t.Fatal("debouncer did not return after context cancellation")
+	}
+}
+
+func TestStartRecreatesObjectAfterDeletionWithoutCancellation(t *testing.T) {
+	previousProcs := runtime.GOMAXPROCS(1)
+	t.Cleanup(func() { runtime.GOMAXPROCS(previousProcs) })
+
+	input := make(chan watch.Event)
+	d, err := New(input, "200ms")
+	if err != nil {
+		t.Fatalf("failed to create debouncer: %s", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	done := make(chan error, 1)
+	go func() { done <- d.Start(ctx) }()
+
+	service := func(eventType watch.EventType, resourceVersion string) watch.Event {
+		return watch.Event{
+			Type: eventType,
+			Object: &v1.Service{ObjectMeta: metav1.ObjectMeta{
+				Name:            "example",
+				Namespace:       "default",
+				ResourceVersion: resourceVersion,
+			}},
+		}
+	}
+
+	send := func(event watch.Event) {
+		t.Helper()
+		select {
+		case input <- event:
+		case <-time.After(time.Second):
+			t.Fatal("debouncer did not receive the test event")
+		}
+	}
+
+	send(service(watch.Added, "initial"))
+	select {
+	case event := <-d.output:
+		if event.Type != watch.Added {
+			t.Fatalf("expected initial Added event, got %s", event.Type)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("debouncer did not emit the initial event")
+	}
+
+	eventNs, exists := d.getNs("default")
+	if !exists {
+		t.Fatal("debouncer did not create the namespace map")
+	}
+	oldObject, exists := eventNs.get("example")
+	if !exists {
+		t.Fatal("debouncer did not create the object")
+	}
+
+	send(service(watch.Deleted, "deleted"))
+	select {
+	case event := <-d.output:
+		if event.Type != watch.Deleted {
+			t.Fatalf("expected Deleted event, got %s", event.Type)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("debouncer did not emit the Deleted event")
+	}
+
+	select {
+	case <-oldObject.stopChan:
+	case <-time.After(time.Second):
+		t.Fatal("object did not self-terminate")
+	}
+	if _, exists := eventNs.get("example"); exists {
+		t.Fatal("self-terminated object remained in the namespace map")
+	}
+
+	send(service(watch.Modified, "fresh"))
+	select {
+	case event := <-d.output:
+		if event.Type != watch.Modified {
+			t.Fatalf("expected fresh Modified event, got %s", event.Type)
+		}
+		service, ok := event.Object.(*v1.Service)
+		if !ok {
+			t.Fatalf("expected a Service event, got %T", event.Object)
+		}
+		if service.ResourceVersion != "fresh" {
+			t.Fatalf("expected the fresh event, got resource version %q", service.ResourceVersion)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("debouncer did not process the fresh event after object deletion")
+	}
+
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("debouncer returned an error: %s", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("debouncer did not stop")
+	}
+}