plugin/dnstap: fix self-deadlock in listener broadcast on client flush error (#8260)

listener.Dnstap holds clientsMu.RLock() while iterating connected sink
clients. In the flush-error branch it called removeClient(c) synchronously,
but removeClient takes clientsMu.Lock(). A sync.RWMutex is not reentrant, so
the goroutine blocks forever waiting to acquire the write lock it can never
get while holding the read lock. The queued Lock() then blocks every
subsequent Dnstap broadcast and close(), and the goroutine leaks.

A flush error is the normal failure mode for a slow or disconnected sink
client (writeMsg buffers into framestream and succeeds; flush does the real
socket write and fails), so a single misbehaving client wedged the whole
listen path. Because Dnstap runs inline in the request-serving goroutine via
TapMessageWithMetadata, this could cascade into stalled request handling.

The write-error branch one line up already offloaded with `go removeClient(c)`.
Do the same in the flush-error branch and drop the early return so the
broadcast still reaches the remaining clients.

Assisted-by: Claude Opus 4.8

Signed-off-by: Pavel Lazureykis <pavel@lazureykis.dev>

plugin/dnstap/listener.go

@@ -183,11 +183,8 @@ func (l *listener) Dnstap(payload *tap.Dnstap) {
 		default:
 			if err := c.enc.writeMsg(payload); err != nil {
 				go l.removeClient(c)
-			} else {
-				if err := c.enc.flush(); err != nil {
-					l.removeClient(c)
-					return
-				}
+			} else if err := c.enc.flush(); err != nil {
+				go l.removeClient(c)
 			}
 		}
 	}

plugin/dnstap/listener_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	tap "github.com/dnstap/golang-dnstap"
+	fs "github.com/farsightsec/golang-framestream"
 )
 
 func TestListenerCreation(t *testing.T) {
@@ -63,6 +64,76 @@ func TestListenerBroadcast(_ *testing.T) {
 	l.Dnstap(testPayload)
 }
 
+// TestListenerDnstapFlushErrorNoDeadlock is a regression test for a self-deadlock:
+// Dnstap held clientsMu.RLock and, when a client's flush() failed, called
+// removeClient inline. removeClient takes clientsMu.Lock, which a non-reentrant
+// RWMutex can never grant to a goroutine already holding the RLock, wedging every
+// future broadcast and close(). A flush failure is the normal case for a slow or
+// disconnected sink client, so the whole listen path must survive it.
+func TestListenerDnstapFlushErrorNoDeadlock(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+
+	// Peer completes the framestream bidirectional handshake so newEncoder succeeds.
+	stop := make(chan struct{})
+	defer close(stop)
+	accepted := make(chan struct{})
+	go func() {
+		conn, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		if _, err := fs.NewDecoder(conn, &fs.DecoderOptions{
+			ContentType:   []byte("protobuf:dnstap.Dnstap"),
+			Bidirectional: true,
+		}); err != nil {
+			return
+		}
+		close(accepted)
+		<-stop
+	}()
+
+	conn, err := net.Dial("tcp", ln.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	enc, err := newEncoder(conn, time.Second)
+	if err != nil {
+		t.Fatalf("newEncoder: %v", err)
+	}
+	<-accepted
+
+	// Break the connection: writeMsg still buffers into framestream successfully,
+	// but flush() (the real socket write) fails, exercising the flush-error branch.
+	conn.Close()
+
+	l := newListener("tcp", "127.0.0.1:0")
+	c := &client{conn: conn, enc: enc, quit: make(chan struct{})}
+	l.clients[c] = struct{}{}
+
+	// Message.Type is a required proto field; without it proto.Marshal fails in
+	// writeMsg and we never reach the flush-error branch this test exercises.
+	dnstapType := tap.Dnstap_MESSAGE
+	msgType := tap.Message_CLIENT_QUERY
+	payload := &tap.Dnstap{Type: &dnstapType, Message: &tap.Message{Type: &msgType}}
+
+	done := make(chan struct{})
+	go func() {
+		l.Dnstap(payload)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Dnstap deadlocked: removeClient takes clientsMu.Lock while Dnstap holds RLock")
+	}
+}
+
 func TestListenerRemoveClient(t *testing.T) {
 	l := newListener("tcp", "127.0.0.1:16002")