From b481ca9cbae92b05fafba2de34265df301edd4b2 Mon Sep 17 00:00:00 2001
From: mark a. foltz <[email protected]>
Date: Tue, 15 Sep 2026 18:35:15 -0700
Subject: [PATCH] [remoting] Improve threading semantics for StreamProvider

StreamProvider owns MediaStream instances which bind Mojo endpoints
and WeakPtrs to the media task runner. When DemuxerManager was destroyed
on the main thread (e.g. during player teardown), it destroyed
StreamProvider inline on the main thread instead of dispatching to
media_task_runner_. This led to Mojo sequence check failures and UAFs.

This CL:
- Updates DemuxerManager to delete StreamProvider
  (kStreamProviderDemuxer) on the media task runner via DeleteSoon().
- Enforces via CHECK that StreamProvider is deleted on the media task
  runner (or main thread in single-threaded test environments).
- Ensures StreamProvider::Stop() invalidates weak pointers and releases
  media resources (audio/video streams) immediately, and is called from
  the destructor if not already stopped.
- Uses base::SequenceChecker in StreamProvider and MediaStream to verify
  calls on the main and media sequences.
- Documents the threading model in StreamProvider.
- Adds unit tests verifying Stop() resource cleanup and media-sequence
  deletion.
- Uses a base::SequenceChecker throughout to validate threading
  assumptions.
- Improves naming of weak pointer factories.

Fixed: 553157579
Change-Id: I70d851199116d69a4400082946b0e1644a18ad4e
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/8394446
Reviewed-by: Jordan Bayles <[email protected]>
Commit-Queue: Jordan Bayles <[email protected]>
Cr-Commit-Position: refs/heads/main@{#1698580}
---
diff --git a/media/filters/demuxer_manager.cc b/media/filters/demuxer_manager.cc
index 7401dbf..bf56240 100644
--- a/media/filters/demuxer_manager.cc
+++ b/media/filters/demuxer_manager.cc
@@ -80,10 +80,19 @@
 }
 
 DemuxerManager::~DemuxerManager() {
-  // ManifestDemuxer and FrameInjectingDemuxer have multiple outstanding weak
-  // pointers bound to the media thread, and need to be deleted there.
+  if (demuxer_override_) {
+    DemuxerType type = demuxer_override_->GetDemuxerType();
+    if (type == DemuxerType::kStreamProviderDemuxer ||
+        type == DemuxerType::kFrameInjectingDemuxer) {
+      media_task_runner_->DeleteSoon(FROM_HERE, std::move(demuxer_override_));
+    }
+  }
+  // ManifestDemuxer, FrameInjectingDemuxer and StreamProvider have multiple
+  // outstanding weak pointers bound to the media thread, and need to be deleted
+  // there.
   if (GetDemuxerType() == DemuxerType::kManifestDemuxer ||
-      GetDemuxerType() == DemuxerType::kFrameInjectingDemuxer) {
+      GetDemuxerType() == DemuxerType::kFrameInjectingDemuxer ||
+      GetDemuxerType() == DemuxerType::kStreamProviderDemuxer) {
     media_task_runner_->DeleteSoon(FROM_HERE, std::move(demuxer_));
   }
 }
diff --git a/media/remoting/end2end_test_renderer.h b/media/remoting/end2end_test_renderer.h
index f70c300..690369a 100644
--- a/media/remoting/end2end_test_renderer.h
+++ b/media/remoting/end2end_test_renderer.h
@@ -88,8 +88,10 @@
   // Receiver components.
   std::unique_ptr<TestRemotee> media_remotee_;
   raw_ptr<ReceiverController> receiver_controller_;
-  std::unique_ptr<Receiver> receiver_;
+  // `stream_provider_` must be declared before `receiver_` so that `receiver_`
+  // (which holds a raw_ptr to `stream_provider_`) is destroyed first.
   std::unique_ptr<StreamProvider> stream_provider_;
+  std::unique_ptr<Receiver> receiver_;
   raw_ptr<openscreen::cast::RpcMessenger> receiver_rpc_messenger_;
 
   // Handle of |receiver_|
diff --git a/media/remoting/stream_provider.cc b/media/remoting/stream_provider.cc
index 49ea1d76..14d4e52b 100644
--- a/media/remoting/stream_provider.cc
+++ b/media/remoting/stream_provider.cc
@@ -64,8 +64,10 @@
       remote_handle_(remote_handle),
       rpc_handle_(rpc_messenger_->GetUniqueHandle()) {
   DCHECK(remote_handle_ != RpcMessenger::kInvalidHandle);
+  DCHECK_CALLED_ON_VALID_SEQUENCE(main_sequence_checker_);
+  DETACH_FROM_SEQUENCE(media_sequence_checker_);
 
-  media_weak_this_ = media_weak_factory_.GetWeakPtr();
+  media_weak_this_ = media_stream_weak_factory_.GetWeakPtr();
 
   auto receive_callback = base::BindPostTask(
       media_task_runner_,
@@ -78,23 +80,23 @@
 }
 
 StreamProvider::MediaStream::~MediaStream() {
-  DCHECK(main_task_runner_->BelongsToCurrentThread());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(main_sequence_checker_);
   rpc_messenger_->UnregisterMessageReceiverCallback(rpc_handle_);
 }
 
 void StreamProvider::MediaStream::Destroy() {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
 
-  // Invalid weak pointers to prevent |this| from receiving RPC calls on the
+  // Invalid weak pointers to prevent `this` from receiving RPC calls on the
   // media thread.
-  media_weak_factory_.InvalidateWeakPtrs();
+  media_stream_weak_factory_.InvalidateWeakPtrs();
 
   // Unbind all mojo pipes and bindings.
   receiver_.reset();
   decoder_buffer_reader_.reset();
 
-  // After invalidating all weak ptrs of |media_weak_factory_|, MediaStream
-  // won't be access anymore, so using |this| here is safe.
+  // After invalidating all weak ptrs of `media_stream_weak_factory_`,
+  // MediaStream won't be access anymore, so using `this` here is safe.
   main_task_runner_->DeleteSoon(FROM_HERE, this);
 }
 
@@ -109,7 +111,7 @@
 
 void StreamProvider::MediaStream::Initialize(
     base::OnceClosure init_done_callback) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   DCHECK(init_done_callback);
 
   if (init_done_callback_) {
@@ -128,7 +130,7 @@
 
 void StreamProvider::MediaStream::InitializeDataPipe(
     mojo::ScopedDataPipeConsumerHandle data_pipe) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
 
   decoder_buffer_reader_ =
       std::make_unique<MojoDecoderBufferReader>(std::move(data_pipe));
@@ -137,7 +139,7 @@
 
 void StreamProvider::MediaStream::ReceiveFrame(uint32_t count,
                                                mojom::DecoderBufferPtr buffer) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   DCHECK(decoder_buffer_reader_);
 
   auto callback = base::BindPostTaskToCurrentDefault(
@@ -147,7 +149,7 @@
 }
 
 void StreamProvider::MediaStream::FlushUntil(uint32_t count) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
 
   if (count < current_frame_count_)
     return;
@@ -170,7 +172,7 @@
 
 void StreamProvider::MediaStream::OnReceivedRpc(
     std::unique_ptr<openscreen::cast::RpcMessage> message) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   DCHECK(message->handle() == rpc_handle_);
 
   switch (message->proc()) {
@@ -187,7 +189,7 @@
 
 void StreamProvider::MediaStream::OnInitializeCallback(
     std::unique_ptr<openscreen::cast::RpcMessage> message) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   const openscreen::cast::DemuxerStreamInitializeCallback callback_message =
       message->demuxerstream_initializecb_rpc();
   if (callback_message.type() != type_) {
@@ -233,7 +235,7 @@
 }
 
 void StreamProvider::MediaStream::CompleteInitialize() {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
 
   // Initialization finished when received RPC_DS_INITIALIZE_CALLBACK and
   // |decoder_buffer_reader_| is created.
@@ -250,7 +252,7 @@
 
 void StreamProvider::MediaStream::OnReadUntilCallback(
     std::unique_ptr<openscreen::cast::RpcMessage> message) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
 
   if (!read_until_sent_) {
     OnError("Unexpected ReadUntilCallback");
@@ -326,7 +328,7 @@
 
 // Only return one buffer at a time so we ignore the count.
 void StreamProvider::MediaStream::Read(uint32_t /*count*/, ReadCB read_cb) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   DCHECK(read_complete_callback_.is_null());
   DCHECK(read_cb);
 
@@ -347,7 +349,7 @@
 }
 
 void StreamProvider::MediaStream::CompleteRead(DemuxerStream::Status status) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
 
   switch (status) {
     case DemuxerStream::kConfigChanged:
@@ -378,11 +380,13 @@
 }
 
 AudioDecoderConfig StreamProvider::MediaStream::audio_decoder_config() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   DCHECK(type_ == DemuxerStream::AUDIO);
   return audio_decoder_config_;
 }
 
 VideoDecoderConfig StreamProvider::MediaStream::video_decoder_config() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   DCHECK(type_ == DemuxerStream::VIDEO);
   return video_decoder_config_;
 }
@@ -402,7 +406,7 @@
 void StreamProvider::MediaStream::AppendBuffer(
     uint32_t count,
     scoped_refptr<DecoderBuffer> buffer) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
 
   // Drop flushed frame.
   if (count < current_frame_count_)
@@ -434,8 +438,10 @@
       rpc_messenger_(receiver_controller_->rpc_messenger()) {
   DCHECK(receiver_controller_);
   DCHECK(rpc_messenger_);
+  DCHECK_CALLED_ON_VALID_SEQUENCE(main_sequence_checker_);
+  DETACH_FROM_SEQUENCE(media_sequence_checker_);
 
-  media_weak_this_ = media_weak_factory_.GetWeakPtr();
+  media_weak_this_ = stream_provider_weak_factory_.GetWeakPtr();
 
   auto callback = base::BindPostTask(
       media_task_runner_,
@@ -449,9 +455,30 @@
 }
 
 StreamProvider::~StreamProvider() {
-  DCHECK(main_task_runner_->BelongsToCurrentThread());
-  rpc_messenger_->UnregisterMessageReceiverCallback(
-      RpcMessenger::kAcquireDemuxerHandle);
+  // Must be destroyed on `media_task_runner_` (or `main_task_runner_` in
+  // single-threaded tests).
+  CHECK(media_task_runner_->RunsTasksInCurrentSequence() &&
+        (!main_task_runner_->BelongsToCurrentThread() ||
+         main_task_runner_ == media_task_runner_));
+
+  if (!stopped_) {
+    Stop();
+  }
+
+  if (rpc_messenger_) {
+    if (main_task_runner_->BelongsToCurrentThread()) {
+      rpc_messenger_->UnregisterMessageReceiverCallback(
+          RpcMessenger::kAcquireDemuxerHandle);
+    } else {
+      // |rpc_messenger_| is owned by |receiver_controller_| which is a
+      // singleton per process, so it's safe to use Unretained() here.
+      main_task_runner_->PostTask(
+          FROM_HERE, base::BindOnce(&openscreen::cast::RpcMessenger::
+                                        UnregisterMessageReceiverCallback,
+                                    base::Unretained(rpc_messenger_),
+                                    RpcMessenger::kAcquireDemuxerHandle));
+    }
+  }
 }
 
 std::string StreamProvider::GetDisplayName() const {
@@ -464,43 +491,70 @@
 
 void StreamProvider::Initialize(DemuxerHost* host,
                                 PipelineStatusCallback status_cb) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   init_done_callback_ = std::move(status_cb);
   CompleteInitialize();
 }
 
-void StreamProvider::AbortPendingReads() {}
+void StreamProvider::AbortPendingReads() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
+}
 
-void StreamProvider::StartWaitingForSeek(base::TimeDelta seek_time) {}
+void StreamProvider::StartWaitingForSeek(base::TimeDelta seek_time) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
+}
 
-void StreamProvider::CancelPendingSeek(base::TimeDelta seek_time) {}
+void StreamProvider::CancelPendingSeek(base::TimeDelta seek_time) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
+}
 
 void StreamProvider::Seek(base::TimeDelta time,
                           PipelineStatusCallback seek_cb) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   media_task_runner_->PostTask(FROM_HERE,
                                base::BindOnce(std::move(seek_cb), PIPELINE_OK));
 }
 
 bool StreamProvider::IsSeekable() const {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   return false;
 }
 
-void StreamProvider::Stop() {}
+void StreamProvider::Stop() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
+  if (stopped_) {
+    return;
+  }
+  stopped_ = true;
+
+  if (init_done_callback_) {
+    std::move(init_done_callback_).Run(PIPELINE_ERROR_ABORT);
+  }
+
+  stream_provider_weak_factory_.InvalidateWeakPtrs();
+
+  audio_stream_.reset();
+  video_stream_.reset();
+}
 
 base::TimeDelta StreamProvider::GetStartTime() const {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   return base::TimeDelta();
 }
 
 base::Time StreamProvider::GetTimelineOffset() const {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   return base::Time();
 }
 
 int64_t StreamProvider::GetMemoryUsage() const {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   return 0;
 }
 
 std::optional<container_names::MediaContainerName>
 StreamProvider::GetContainerForMetrics() const {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   return std::optional<container_names::MediaContainerName>();
 }
 
@@ -508,30 +562,14 @@
                                      std::optional<MediaTrack::Id> track_id,
                                      base::TimeDelta curr_time,
                                      TrackChangeCB change_completed_cb) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   std::move(change_completed_cb).Run(nullptr);
   DVLOG(1) << "Track changes are not supported.";
 }
 
-void StreamProvider::Destroy() {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
-
-  if (init_done_callback_)
-    std::move(init_done_callback_).Run(PIPELINE_ERROR_ABORT);
-
-  // Invalid weak pointers to prevent |this| from receiving RPC calls on the
-  // media thread.
-  media_weak_factory_.InvalidateWeakPtrs();
-
-  audio_stream_.reset();
-  video_stream_.reset();
-
-  // After invalidating all weak ptrs of |media_weak_factory_|, StreamProvider
-  // won't be access anymore, so using |this| here is safe.
-  main_task_runner_->DeleteSoon(FROM_HERE, this);
-}
-
 void StreamProvider::OnReceivedRpc(
     std::unique_ptr<openscreen::cast::RpcMessage> message) {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
   switch (message->proc()) {
     case openscreen::cast::RpcMessage::RPC_ACQUIRE_DEMUXER:
       OnAcquireDemuxer(std::move(message));
@@ -543,7 +581,10 @@
 
 void StreamProvider::OnAcquireDemuxer(
     std::unique_ptr<openscreen::cast::RpcMessage> message) {
-  DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK_CALLED_ON_VALID_SEQUENCE(media_sequence_checker_);
+  if (stopped_) {
+    return;
... (file diff truncated)

diff --git a/media/remoting/stream_provider_unittest.cc b/media/remoting/stream_provider_unittest.cc
index 84d0cff..9d125289 100644
--- a/media/remoting/stream_provider_unittest.cc
+++ b/media/remoting/stream_provider_unittest.cc
@@ -7,7 +7,9 @@
 #include "base/functional/callback_helpers.h"
 #include "base/memory/raw_ptr.h"
 #include "base/memory/scoped_refptr.h"
+#include "base/run_loop.h"
 #include "base/task/single_thread_task_runner.h"
+#include "base/test/run_until.h"
 #include "base/test/task_environment.h"
 #include "media/base/audio_decoder_config.h"
 #include "media/base/demuxer_stream.h"
@@ -355,5 +357,51 @@
   task_environment_.RunUntilIdle();
 }
 
+TEST_F(StreamProviderTest, StopReleasesResourcesAndInvalidatesCallbacks) {
+  InitializeDemuxer();
+  SendRpcAcquireDemuxer();
+  EXPECT_TRUE(
+      base::test::RunUntil([&]() { return stream_provider_initialized_; }));
+  EXPECT_EQ(2u, stream_provider_->GetAllStreams().size());
+
+  stream_provider_->Stop();
+  // Clear unowned pointers in the test fixture since the streams were released.
+  audio_stream_ = nullptr;
+  video_stream_ = nullptr;
+  // Streams should be released immediately.
+  EXPECT_TRUE(stream_provider_->GetAllStreams().empty());
+
+  // Further RPC messages should be ignored.
+  SendRpcAcquireDemuxer();
+  base::RunLoop run_loop;
+  base::SingleThreadTaskRunner::GetCurrentDefault()->PostTask(
+      FROM_HERE, run_loop.QuitClosure());
+  run_loop.Run();
+  EXPECT_TRUE(stream_provider_->GetAllStreams().empty());
+
+  // Calling Stop() again should be a safe no-op.
+  stream_provider_->Stop();
+}
+
+TEST_F(StreamProviderTest, StopAbortsPendingInitialize) {
+  PipelineStatus status = PIPELINE_OK;
+  stream_provider_->Initialize(
+      nullptr, base::BindOnce([](PipelineStatus* out_status,
+                                 PipelineStatus s) { *out_status = s; },
+                              &status));
+  stream_provider_->Stop();
+  EXPECT_EQ(PIPELINE_ERROR_ABORT, status);
+}
+
+TEST_F(StreamProviderTest, DestructorCallsStopIfNotStopped) {
+  PipelineStatus status = PIPELINE_OK;
+  stream_provider_->Initialize(
+      nullptr, base::BindOnce([](PipelineStatus* out_status,
+                                 PipelineStatus s) { *out_status = s; },
+                              &status));
+  stream_provider_.reset();
+  EXPECT_EQ(PIPELINE_ERROR_ABORT, status);
+}
+
 }  // namespace remoting
 }  // namespace media