[audio_core] Fix capture overflow busy-spin race In BaseCapturer::Process(), when the CapturePacketQueue (CPQ) is empty during async capture, the mix thread waits for packets in WaitForPendingPacket(). If the FIDL thread shuts down async capture (via StopAsyncCapture or BeginShutdown), CPQ::Shutdown() unblocks WaitForPendingPacket(). Previously, WaitForPendingPacket() returned void, and Process() only checked `state_.load() == State::Shutdown`. If the capturer had not yet transitioned to State::Shutdown (e.g. it was still in State::AsyncOperating or State::AsyncStopping), Process() assumed a new packet arrived, reported an overflow, and continued looping. Because CPQ was shutdown, NextMixerJob() returned nullopt and WaitForPendingPacket() returned immediately with 0 delay, causing Process() to busy-spin and invoke ReportOverflow() dozens of times every millisecond. This CL fixes the race by: 1. Updating CapturePacketQueue::WaitForPendingPacket() to return bool (`!shutdown_ && !pending_.empty()`), indicating whether a packet is actually available to mix. 2. In BaseCapturer::Process(), checking WaitForPendingPacket()'s return value, as well as `state_ == State::AsyncOperating`. If the queue was shut down or the capturer is stopping, Process() exits immediately with ZX_OK rather than reporting false overflows. 3. Adding unit test coverage for CPQ shutdown/wait unblocking and capturer async stop during packet starvation.. New/updated unittest cases in audio_core_unittests: CapturePacketQueueTest: WaitForPendingPacketWhenAlreadyShutdown WaitForPendingPacketReturnsTrueWhenPacketAvailable WaitForPendingPacketReturnsTrueWhenPacketPushed WaitForPendingPacketReturnsTrueWhenPacketRecycled WaitForPendingPacketReturnsFalseOnShutdown AudioCapturerTest: StopAsyncCaptureCleanShutdown Bug: 537405213 Bug: 539867284 Fixed: 545709661 Test: audio_core_unittests CONV: dd9b809f-fe9e-46c8-b4da-dd8f8af98fcd TAG: agy Change-Id: I3eb48773f2acad6dbe72f57299101b4a44990a5b Reviewed-on: https://fuchsia-review.googlesource.com/c/fuchsia/+/1758453 Fuchsia-Auto-Submit: Martin Puryear <mpuryear@google.com> Commit-Queue: Martin Puryear <mpuryear@google.com> Reviewed-by: Andres Oportus <andresoportus@google.com>

diff --git a/src/media/audio/audio_core/audio_capturer_unittest.cc b/src/media/audio/audio_core/audio_capturer_unittest.cc
index 030ac57..ff2dcb5 100644
--- a/src/media/audio/audio_core/audio_capturer_unittest.cc
+++ b/src/media/audio/audio_core/audio_capturer_unittest.cc

@@ -378,5 +378,31 @@
   EXPECT_TRUE(channel_dropped);
 }
 
+// Verify that stopping async capture when all packets are in flight / waiting completes cleanly,
+// invoking the stop callback and emitting OnEndOfStream without spinning or hanging.
+TEST_F(AudioCapturerTest, StopAsyncCaptureCleanShutdown) {
+  zx::vmo duplicate;
+  ASSERT_EQ(
+      vmo_.duplicate(ZX_RIGHT_TRANSFER | ZX_RIGHT_WRITE | ZX_RIGHT_READ | ZX_RIGHT_MAP, &duplicate),
+      ZX_OK);
+  fidl_capturer_->AddPayloadBuffer(0, std::move(duplicate));
+  RunLoopUntilIdle();
+
+  fidl_capturer_->StartAsyncCapture(480);
+  RunLoopUntilIdle();
+
+  bool stop_callback_received = false;
+  bool end_of_stream_received = false;
+  fidl_capturer_.events().OnEndOfStream = [&end_of_stream_received]() {
+    end_of_stream_received = true;
+  };
+
+  fidl_capturer_->StopAsyncCapture([&stop_callback_received]() { stop_callback_received = true; });
+  RunLoopUntilIdle();
+
+  EXPECT_TRUE(stop_callback_received);
+  EXPECT_TRUE(end_of_stream_received);
+}
+
 }  // namespace
 }  // namespace media::audio

diff --git a/src/media/audio/audio_core/base_capturer.cc b/src/media/audio/audio_core/base_capturer.cc
index 1770650..bf0eebf 100644
--- a/src/media/audio/audio_core/base_capturer.cc
+++ b/src/media/audio/audio_core/base_capturer.cc

@@ -591,8 +591,12 @@
       // Wait until we have another packet or have shut down.
       // This waits for the caller to ACK a packet, so it might block indefinitely.
       auto overflow_start = zx::clock::get_monotonic();
-      pq->WaitForPendingPacket();
-      if (state_.load() == State::Shutdown) {
+      bool has_packet = pq->WaitForPendingPacket();
+
+      // If the queue was shut down, or if the capturer is no longer in AsyncOperating mode
+      // (e.g. StopAsyncCapture, Shutdown, or Disconnect occurred while we were waiting),
+      // exit immediately without reporting a spurious overflow or looping.
+      if (!has_packet || state_.load() != State::AsyncOperating) {
         return ZX_OK;
       }
 

diff --git a/src/media/audio/audio_core/capture_packet_queue.cc b/src/media/audio/audio_core/capture_packet_queue.cc
index 5e54f141..8736f61 100644
--- a/src/media/audio/audio_core/capture_packet_queue.cc
+++ b/src/media/audio/audio_core/capture_packet_queue.cc

@@ -254,12 +254,13 @@
 };
 }  // namespace
 
-void CapturePacketQueue::WaitForPendingPacket() {
+bool CapturePacketQueue::WaitForPendingPacket() {
   TRACE_DURATION("audio", "CapturePacketQueue::WaitForPendingPacket");
   scoped_unique_lock lock(mutex_);
   while (!shutdown_ && pending_.empty()) {
     pending_signal_.wait(lock);
   }
+  return !shutdown_ && !pending_.empty();
 }
 
 }  // namespace media::audio

diff --git a/src/media/audio/audio_core/capture_packet_queue.h b/src/media/audio/audio_core/capture_packet_queue.h
index 36d3dae..5dab5cb 100644
--- a/src/media/audio/audio_core/capture_packet_queue.h
+++ b/src/media/audio/audio_core/capture_packet_queue.h

@@ -182,7 +182,8 @@
   void Shutdown();
 
   // Block until the pending queue is non-empty or the queue has been shut down.
-  void WaitForPendingPacket();
+  // Returns true if a pending packet is available to mix, or false if the queue was shut down.
+  bool WaitForPendingPacket();
 
  private:
   enum class Mode { Preallocated, DynamicallyAllocated };

diff --git a/src/media/audio/audio_core/capture_packet_queue_unittest.cc b/src/media/audio/audio_core/capture_packet_queue_unittest.cc
index b71a47c..c38d860 100644
--- a/src/media/audio/audio_core/capture_packet_queue_unittest.cc
+++ b/src/media/audio/audio_core/capture_packet_queue_unittest.cc

@@ -6,6 +6,9 @@
 
 #include <lib/syslog/cpp/macros.h>
 
+#include <atomic>
+#include <thread>
+
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
@@ -390,13 +393,138 @@
   ASSERT_TRUE(push_result.is_error());
 }
 
-// Verify that WaitForPendingPacket returns immediately on shutdown: no block or double-unlock.
-TEST_F(CapturePacketQueueTest, WaitForPendingPacketShutdown) {
+// Verify that WaitForPendingPacket returns immediately with false when the queue is already shut
+// down (exercising the non-blocking shutdown path where NextMixerJob returns nullopt).
+TEST_F(CapturePacketQueueTest, WaitForPendingPacketWhenAlreadyShutdown) {
   CreateMapper(50);
   auto pq = CapturePacketQueue::CreateDynamicallyAllocated(payload_buffer_, kFormat);
   pq->Shutdown();
 
-  pq->WaitForPendingPacket();
+  EXPECT_FALSE(pq->WaitForPendingPacket());
+  EXPECT_EQ(pq->NextMixerJob(), std::nullopt);
+}
+
+// Verify that WaitForPendingPacket returns true immediately without blocking if pending packets
+// are already available in the queue.
+TEST_F(CapturePacketQueueTest, WaitForPendingPacketReturnsTrueWhenPacketAvailable) {
+  CreateMapper(50);
+  auto pq = CapturePacketQueue::CreateDynamicallyAllocated(payload_buffer_, kFormat);
+  auto push_result = pq->PushPending(0, 10, nullptr);
+  ASSERT_TRUE(push_result.is_ok());
+
+  EXPECT_TRUE(pq->WaitForPendingPacket());
+  auto mix_state = pq->NextMixerJob();
+  ASSERT_TRUE(mix_state.has_value());
+  EXPECT_EQ(mix_state->frames, 10u);
+}
+
+// Verify that WaitForPendingPacket blocks across threads when the queue is empty, and returns true
+// once a new packet is dynamically pushed via PushPending.
+TEST_F(CapturePacketQueueTest, WaitForPendingPacketReturnsTrueWhenPacketPushed) {
+  CreateMapper(50);
+  auto pq = CapturePacketQueue::CreateDynamicallyAllocated(payload_buffer_, kFormat);
+
+  std::atomic<bool> worker_started = false;
+  std::atomic<bool> wait_finished = false;
+  std::atomic<bool> wait_result = false;
+  std::thread worker([&pq, &worker_started, &wait_finished, &wait_result]() {
+    worker_started = true;
+    wait_result = pq->WaitForPendingPacket();
+    wait_finished = true;
+  });
+
+  while (!worker_started.load()) {
+    std::this_thread::yield();
+  }
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+  EXPECT_FALSE(wait_finished.load());
+  auto push_result = pq->PushPending(0, 10, nullptr);
+  ASSERT_TRUE(push_result.is_ok());
+
+  worker.join();
+  EXPECT_TRUE(wait_finished.load());
+  EXPECT_TRUE(wait_result.load());
+
+  auto mix_state = pq->NextMixerJob();
+  ASSERT_TRUE(mix_state.has_value());
+  EXPECT_EQ(mix_state->frames, 10u);
+}
+
+// Verify that WaitForPendingPacket blocks in preallocated mode when all packets are in flight,
+// and unblocks returning true once a packet is returned via Recycle (exercising async capture
+// overflow recovery).
+TEST_F(CapturePacketQueueTest, WaitForPendingPacketReturnsTrueWhenPacketRecycled) {
+  CreateMapper(20);
+  auto result = CapturePacketQueue::CreatePreallocated(payload_buffer_, kFormat, 10);
+  ASSERT_TRUE(result.is_ok()) << result.error();
+  auto pq = result.take_value();
+
+  // Pop both packets so the pending queue is exhausted.
+  auto mix_state1 = pq->NextMixerJob().value();
+  ASSERT_EQ(CapturePacketQueue::PacketMixStatus::Done, pq->FinishMixerJob(mix_state1));
+  auto p1 = pq->PopReady();
+
+  auto mix_state2 = pq->NextMixerJob().value();
+  ASSERT_EQ(CapturePacketQueue::PacketMixStatus::Done, pq->FinishMixerJob(mix_state2));
+  auto p2 = pq->PopReady();
+
+  ASSERT_EQ(pq->PendingSize(), 0u);
+
+  std::atomic<bool> worker_started = false;
+  std::atomic<bool> wait_finished = false;
+  std::atomic<bool> wait_result = false;
+  std::thread worker([&pq, &worker_started, &wait_finished, &wait_result]() {
+    worker_started = true;
+    wait_result = pq->WaitForPendingPacket();
+    wait_finished = true;
+  });
+
+  while (!worker_started.load()) {
+    std::this_thread::yield();
+  }
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+  EXPECT_FALSE(wait_finished.load());
+  auto recycle_result = pq->Recycle(p1->stream_packet());
+  ASSERT_TRUE(recycle_result.is_ok()) << recycle_result.error();
+
+  worker.join();
+  EXPECT_TRUE(wait_finished.load());
+  EXPECT_TRUE(wait_result.load());
+
+  auto mix_state = pq->NextMixerJob();
+  ASSERT_TRUE(mix_state.has_value());
+  EXPECT_EQ(mix_state->frames, 10u);
+}
+
+// Verify that a thread blocked in WaitForPendingPacket is unblocked and returns false when
+// Shutdown is called concurrently (exercising cancellation/shutdown during packet starvation).
+TEST_F(CapturePacketQueueTest, WaitForPendingPacketReturnsFalseOnShutdown) {
+  CreateMapper(50);
+  auto pq = CapturePacketQueue::CreateDynamicallyAllocated(payload_buffer_, kFormat);
+
+  std::atomic<bool> worker_started = false;
+  std::atomic<bool> wait_finished = false;
+  std::atomic<bool> wait_result = true;
+  std::thread worker([&pq, &worker_started, &wait_finished, &wait_result]() {
+    worker_started = true;
+    wait_result = pq->WaitForPendingPacket();
+    wait_finished = true;
+  });
+
+  while (!worker_started.load()) {
+    std::this_thread::yield();
+  }
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  EXPECT_FALSE(wait_finished.load());
+
+  pq->Shutdown();
+  worker.join();
+
+  EXPECT_TRUE(wait_finished.load());
+  EXPECT_FALSE(wait_result.load());
+  EXPECT_EQ(pq->NextMixerJob(), std::nullopt);
 }
 
 }  // namespace media::audio