From c66714ce42bb62b090eeb6fc6f547423ca80d960 Mon Sep 17 00:00:00 2001 From: Viktor Liu Date: Tue, 22 Sep 2026 14:53:41 +0200 Subject: [PATCH] Evict a stalled packet capture without holding the daemon mutex --- client/server/capture.go | 45 +++++++--- client/server/capture_evict_test.go | 129 ++++++++++++++++++++++++++++ 2 files changed, 160 insertions(+), 14 deletions(-) create mode 100644 client/server/capture_evict_test.go diff --git a/client/server/capture.go b/client/server/capture.go index 0b1eaa8e8..327ee9776 100644 --- a/client/server/capture.go +++ b/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() } } diff --git a/client/server/capture_evict_test.go b/client/server/capture_evict_test.go new file mode 100644 index 000000000..d27306d2c --- /dev/null +++ b/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") + } +}