Evict a stalled packet capture without holding the daemon mutex

client/server/capture.go

@@ -186,11 +186,15 @@ func streamToGRPC(r io.Reader, stream proto.DaemonService_StartCaptureServer) er
 // never called (e.g. CLI crash).
 func (s *Server) StartBundleCapture(_ context.Context, req *proto.StartBundleCaptureRequest) (*proto.StartBundleCaptureResponse, error) {
 	s.mutex.Lock()
+	// Registered before the unlock so it runs after it: the eviction teardown
+	// waits on the evicted session's writer and must not hold s.mutex.
+	stopEvicted := func() {}
+	defer func() { stopEvicted() }()
 	defer s.mutex.Unlock()
 
 	s.stopBundleCaptureLocked()
 	s.cleanupBundleCapture()
-	s.evictActiveCaptureLocked()
+	stopEvicted = s.evictActiveCaptureLocked()
 
 	engine, err := s.getCaptureEngineLocked()
 	if err != nil {
@@ -307,28 +311,36 @@ func (s *Server) cleanupBundleCapture() {
 // and a bundle capture is just informational state.
 func (s *Server) claimCapture(sess *capture.Session, cancel func()) (*internal.Engine, error) {
 	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
-	s.evictActiveCaptureLocked()
+	stopEvicted := s.evictActiveCaptureLocked()
 	engine, err := s.getCaptureEngineLocked()
+	if err == nil {
+		s.activeCapture = sess
+		s.activeCaptureCancel = cancel
+	}
+	s.mutex.Unlock()
+
+	stopEvicted()
+
 	if err != nil {
 		return nil, err
 	}
-	s.activeCapture = sess
-	s.activeCaptureCancel = cancel
 	return engine, nil
 }
 
-// evictActiveCaptureLocked tears down whatever capture currently owns
-// the engine slot so a fresh claim can succeed. Caller must hold mutex.
-func (s *Server) evictActiveCaptureLocked() {
+// evictActiveCaptureLocked releases the engine's capture slot from whatever
+// capture currently owns it so a fresh claim can succeed, and returns the rest
+// of that teardown as a function. The returned function is never nil, is safe
+// to call more than once, and blocks until the evicted session's writer
+// goroutine has exited, so the caller must run it only after releasing
+// s.mutex. Caller must hold mutex.
+func (s *Server) evictActiveCaptureLocked() func() {
 	if s.activeCapture == nil {
-		return
+		return func() {}
 	}
 	if s.bundleCapture != nil && s.bundleCapture.sess == s.activeCapture {
 		log.Infof("evicting running bundle capture to start a new capture")
 		s.stopBundleCaptureLocked()
-		return
+		return func() {}
 	}
 	log.Infof("evicting previous streaming capture to start a new one")
 	prev := s.activeCapture
@@ -340,9 +352,14 @@ func (s *Server) evictActiveCaptureLocked() {
 	}
 	s.activeCapture = nil
 	s.activeCaptureCancel = nil
-	prev.Stop()
-	if cancel != nil {
-		cancel()
+	return func() {
+		// Close the output pipe before waiting: Stop waits for the writer
+		// goroutine, which stays blocked in a write for as long as the reader
+		// side is open and not draining.
+		if cancel != nil {
+			cancel()
+		}
+		prev.Stop()
 	}
 }
 

client/server/capture_evict_test.go

@@ -0,0 +1,129 @@
+package server
+
+import (
+	"io"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/netbirdio/netbird/util/capture"
+)
+
+// signalOnWrite delegates to w and closes signal as the first write starts, so
+// a test can wait until the capture's writer goroutine is actually inside a
+// write that will not return.
+type signalOnWrite struct {
+	w      io.Writer
+	once   sync.Once
+	signal chan struct{}
+}
+
+func (s *signalOnWrite) Write(p []byte) (int, error) {
+	s.once.Do(func() { close(s.signal) })
+	return s.w.Write(p)
+}
+
+// stalledCapture returns a running capture session whose writer goroutine is
+// blocked writing into a pipe nobody reads, plus the cancel that closes the
+// pipe's write end (the shape StartCapture installs).
+func stalledCapture(t *testing.T) (*capture.Session, func()) {
+	t.Helper()
+
+	pr, pw := io.Pipe()
+	t.Cleanup(func() { _ = pr.Close() })
+
+	writing := make(chan struct{})
+	sess, err := capture.NewSession(capture.Options{
+		TextOutput: &signalOnWrite{w: pw, signal: writing},
+		BufSize:    16,
+	})
+	require.NoError(t, err)
+
+	sess.Offer([]byte{0x45, 0x00, 0x00, 0x14}, true)
+
+	select {
+	case <-writing:
+	case <-time.After(5 * time.Second):
+		t.Fatal("capture writer never reached the stalled pipe")
+	}
+
+	return sess, func() { _ = pw.Close() }
+}
+
+// TestClaimCapture_EvictionDoesNotHoldMutex covers the eviction deadlock: the
+// evicted session's Stop waits for a writer goroutine that only the pipe close
+// can release, so doing that wait under s.mutex wedges every RPC that needs the
+// lock. The eviction must close the pipe first and wait outside the lock.
+func TestClaimCapture_EvictionDoesNotHoldMutex(t *testing.T) {
+	sess, cancel := stalledCapture(t)
+
+	s := &Server{}
+	s.activeCapture = sess
+	s.activeCaptureCancel = cancel
+
+	claimed := make(chan struct{})
+	go func() {
+		defer close(claimed)
+		// No engine is wired up, so the claim itself fails. The eviction still
+		// runs, and that is what must not block or strand the mutex.
+		_, _ = s.claimCapture(nil, nil)
+	}()
+
+	select {
+	case <-claimed:
+	case <-time.After(10 * time.Second):
+		t.Fatal("claimCapture blocked evicting a stalled capture")
+	}
+
+	locked := make(chan struct{})
+	go func() {
+		s.mutex.Lock()
+		s.mutex.Unlock()
+		close(locked)
+	}()
+
+	select {
+	case <-locked:
+	case <-time.After(5 * time.Second):
+		t.Fatal("s.mutex still held after evicting a stalled capture")
+	}
+
+	select {
+	case <-sess.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("evicted capture session was never stopped")
+	}
+}
+
+// TestStartBundleCapture_EvictionDoesNotHoldMutex covers the other caller of
+// evictActiveCaptureLocked: a debug bundle started while a streaming capture is
+// stalled must not wedge the daemon either.
+func TestStartBundleCapture_EvictionDoesNotHoldMutex(t *testing.T) {
+	sess, cancel := stalledCapture(t)
+
+	s := &Server{}
+	s.activeCapture = sess
+	s.activeCaptureCancel = cancel
+
+	started := make(chan struct{})
+	go func() {
+		defer close(started)
+		// getCaptureEngineLocked fails without a connected client, which returns
+		// success with capture skipped; the eviction has already happened.
+		_, _ = s.StartBundleCapture(t.Context(), nil)
+	}()
+
+	select {
+	case <-started:
+	case <-time.After(10 * time.Second):
+		t.Fatal("StartBundleCapture blocked evicting a stalled capture")
+	}
+
+	select {
+	case <-sess.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("evicted capture session was never stopped")
+	}
+}