fix(destination): prevent task leak in federated service watchers (#14693)

When unsubscribing a stream from a federated service, we:

1. For each cluster, send an update on the stream's channel to remove endpoints.
2. Call synchronizedGetStream.Stop() so that it stops processing updates.

These steps can race and deadlock: if Stop() has been called before endoint
updates are processed, the subsequent Send() calls block forever.

This change improves test coverage to cover this case as well as that fixed in
f4e6795.

Author: olix0r

controller/api/destination/syncronized_get_stream.go

@@ -2,6 +2,7 @@ package destination
 
 import (
 	"context"
+	"errors"
 
 	pb "github.com/linkerd/linkerd2-proxy-api/go/destination"
 	logging "github.com/sirupsen/logrus"
@@ -21,6 +22,8 @@ type synchronizedGetStream struct {
 	log   *logging.Entry
 }
 
+var errStreamStopped = errors.New("synchronized stream stopped")
+
 func newSyncronizedGetStream(stream pb.Destination_GetServer, log *logging.Entry) *synchronizedGetStream {
 	return &synchronizedGetStream{
 		done:  make(chan struct{}),
@@ -50,8 +53,12 @@ func (s *synchronizedGetStream) RecvMsg(m any) error {
 }
 
 func (s *synchronizedGetStream) Send(update *pb.Update) error {
-	s.ch <- update
-	return nil
+	select {
+	case <-s.done:
+		return errStreamStopped
+	case s.ch <- update:
+		return nil
+	}
 }
 
 func (s *synchronizedGetStream) Start() {

controller/api/destination/syncronized_get_stream_test.go

@@ -0,0 +1,108 @@
+package destination
+
+import (
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	pb "github.com/linkerd/linkerd2-proxy-api/go/destination"
+	"github.com/linkerd/linkerd2/controller/api/util"
+	logging "github.com/sirupsen/logrus"
+)
+
+type blockingDestinationGetServer struct {
+	util.MockServerStream
+	block      chan struct{}
+	sendCalled chan struct{}
+	once       sync.Once
+}
+
+func newBlockingDestinationGetServer() *blockingDestinationGetServer {
+	return &blockingDestinationGetServer{
+		block:      make(chan struct{}),
+		sendCalled: make(chan struct{}),
+	}
+}
+
+func (b *blockingDestinationGetServer) Send(update *pb.Update) error {
+	b.once.Do(func() {
+		close(b.sendCalled)
+	})
+	<-b.block
+	return nil
+}
+
+func (b *blockingDestinationGetServer) unblock() {
+	close(b.block)
+}
+
+// TestSynchronizedGetStreamSendAfterStop ensures Send returns promptly once the
+// stream has been stopped so goroutines don't leak waiting on an unconsumed
+// channel send.
+func TestSynchronizedGetStreamSendAfterStop(t *testing.T) {
+	mock := &mockDestinationGetServer{
+		updatesReceived: make(chan *pb.Update, 1),
+	}
+	stream := newSyncronizedGetStream(mock, logging.WithField("test", t.Name()))
+	stream.Start()
+	stream.Stop()
+
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- stream.Send(&pb.Update{})
+	}()
+
+	select {
+	case err := <-errCh:
+		if !errors.Is(err, errStreamStopped) {
+			t.Fatalf("expected errStreamStopped, got %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Send blocked after Stop")
+	}
+}
+
+func TestSynchronizedGetStreamStopWhileInnerSendBlocked(t *testing.T) {
+	mock := newBlockingDestinationGetServer()
+	stream := newSyncronizedGetStream(mock, logging.WithField("test", t.Name()))
+	stream.Start()
+
+	firstSend := make(chan error, 1)
+	go func() {
+		firstSend <- stream.Send(&pb.Update{})
+	}()
+
+	select {
+	case err := <-firstSend:
+		if err != nil {
+			t.Fatalf("unexpected error from initial Send: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("initial Send did not complete")
+	}
+
+	select {
+	case <-mock.sendCalled:
+	case <-time.After(time.Second):
+		t.Fatal("inner Send was not invoked")
+	}
+
+	stream.Stop()
+
+	secondSend := make(chan error, 1)
+	go func() {
+		secondSend <- stream.Send(&pb.Update{})
+	}()
+
+	select {
+	case err := <-secondSend:
+		if !errors.Is(err, errStreamStopped) {
+			t.Fatalf("expected errStreamStopped, got %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("second Send blocked after Stop")
+	}
+
+	mock.unblock()
+}