diff --git a/call/BUILD.gn b/call/BUILD.gn
index 38b2f41..9011094 100644
--- a/call/BUILD.gn
+++ b/call/BUILD.gn
@@ -228,6 +228,7 @@
     "../logging:rtc_event_rtp_rtcp",
     "../logging:rtc_event_video",
     "../logging:rtc_stream_config",
+    "../modules:module_api",
     "../modules/bitrate_controller",
     "../modules/congestion_controller",
     "../modules/pacing",
@@ -300,7 +301,6 @@
   deps = [
     ":call_interfaces",
     "../api:simulated_network_api",
-    "../modules:module_api",
   ]
 }
 
@@ -317,7 +317,6 @@
     "../api:libjingle_peerconnection_api",
     "../api:simulated_network_api",
     "../api:transport_api",
-    "../modules:module_api",
     "../modules/utility",
     "../rtc_base:checks",
     "../rtc_base:rtc_base_approved",
diff --git a/call/degraded_call.cc b/call/degraded_call.cc
index c066d5d..4b6d1af 100644
--- a/call/degraded_call.cc
+++ b/call/degraded_call.cc
@@ -15,6 +15,55 @@
 #include "rtc_base/location.h"
 
 namespace webrtc {
+
+namespace {
+constexpr int64_t kDoNothingProcessIntervalMs = 5000;
+}  // namespace
+
+FakeNetworkPipeModule::~FakeNetworkPipeModule() = default;
+
+FakeNetworkPipeModule::FakeNetworkPipeModule(
+    Clock* clock,
+    std::unique_ptr<NetworkBehaviorInterface> network_behavior,
+    Transport* transport)
+    : pipe_(clock, std::move(network_behavior), transport) {}
+
+void FakeNetworkPipeModule::SendRtp(const uint8_t* packet,
+                                    size_t length,
+                                    const PacketOptions& options) {
+  pipe_.SendRtp(packet, length, options);
+  MaybeResumeProcess();
+}
+
+void FakeNetworkPipeModule::SendRtcp(const uint8_t* packet, size_t length) {
+  pipe_.SendRtcp(packet, length);
+  MaybeResumeProcess();
+}
+
+void FakeNetworkPipeModule::MaybeResumeProcess() {
+  rtc::CritScope cs(&process_thread_lock_);
+  if (!pending_process_ && pipe_.TimeUntilNextProcess() && process_thread_) {
+    process_thread_->WakeUp(nullptr);
+  }
+}
+
+int64_t FakeNetworkPipeModule::TimeUntilNextProcess() {
+  auto delay = pipe_.TimeUntilNextProcess();
+  rtc::CritScope cs(&process_thread_lock_);
+  pending_process_ = delay.has_value();
+  return delay.value_or(kDoNothingProcessIntervalMs);
+}
+
+void FakeNetworkPipeModule::ProcessThreadAttached(
+    ProcessThread* process_thread) {
+  rtc::CritScope cs(&process_thread_lock_);
+  process_thread_ = process_thread;
+}
+
+void FakeNetworkPipeModule::Process() {
+  pipe_.Process();
+}
+
 DegradedCall::DegradedCall(
     std::unique_ptr<Call> call,
     absl::optional<BuiltInNetworkBehaviorConfig> send_config,
@@ -72,8 +121,8 @@
   if (send_config_ && !send_pipe_) {
     auto network = absl::make_unique<SimulatedNetwork>(*send_config_);
     send_simulated_network_ = network.get();
-    send_pipe_ = absl::make_unique<FakeNetworkPipe>(clock_, std::move(network),
-                                                    config.send_transport);
+    send_pipe_ = absl::make_unique<FakeNetworkPipeModule>(
+        clock_, std::move(network), config.send_transport);
     config.send_transport = this;
     send_process_thread_->RegisterModule(send_pipe_.get(), RTC_FROM_HERE);
   }
@@ -89,8 +138,8 @@
   if (send_config_ && !send_pipe_) {
     auto network = absl::make_unique<SimulatedNetwork>(*send_config_);
     send_simulated_network_ = network.get();
-    send_pipe_ = absl::make_unique<FakeNetworkPipe>(clock_, std::move(network),
-                                                    config.send_transport);
+    send_pipe_ = absl::make_unique<FakeNetworkPipeModule>(
+        clock_, std::move(network), config.send_transport);
     config.send_transport = this;
     send_process_thread_->RegisterModule(send_pipe_.get(), RTC_FROM_HERE);
   }
diff --git a/call/degraded_call.h b/call/degraded_call.h
index 8f062c3..89eafdb 100644
--- a/call/degraded_call.h
+++ b/call/degraded_call.h
@@ -32,6 +32,7 @@
 #include "call/simulated_network.h"
 #include "call/video_receive_stream.h"
 #include "call/video_send_stream.h"
+#include "modules/include/module.h"
 #include "modules/utility/include/process_thread.h"
 #include "rtc_base/bitrate_allocation_strategy.h"
 #include "rtc_base/copy_on_write_buffer.h"
@@ -39,6 +40,30 @@
 #include "system_wrappers/include/clock.h"
 
 namespace webrtc {
+class FakeNetworkPipeModule : public Module {
+ public:
+  FakeNetworkPipeModule(
+      Clock* clock,
+      std::unique_ptr<NetworkBehaviorInterface> network_behavior,
+      Transport* transport);
+  ~FakeNetworkPipeModule() override;
+  void SendRtp(const uint8_t* packet,
+               size_t length,
+               const PacketOptions& options);
+  void SendRtcp(const uint8_t* packet, size_t length);
+
+  // Implements Module interface
+  int64_t TimeUntilNextProcess() override;
+  void ProcessThreadAttached(ProcessThread* process_thread) override;
+  void Process() override;
+
+ private:
+  void MaybeResumeProcess();
+  FakeNetworkPipe pipe_;
+  rtc::CriticalSection process_thread_lock_;
+  ProcessThread* process_thread_ RTC_GUARDED_BY(process_thread_lock_) = nullptr;
+  bool pending_process_ RTC_GUARDED_BY(process_thread_lock_) = false;
+};
 
 class DegradedCall : public Call, private Transport, private PacketReceiver {
  public:
@@ -111,7 +136,7 @@
   const absl::optional<BuiltInNetworkBehaviorConfig> send_config_;
   const std::unique_ptr<ProcessThread> send_process_thread_;
   SimulatedNetwork* send_simulated_network_;
-  std::unique_ptr<FakeNetworkPipe> send_pipe_;
+  std::unique_ptr<FakeNetworkPipeModule> send_pipe_;
   size_t num_send_streams_;
 
   const absl::optional<BuiltInNetworkBehaviorConfig> receive_config_;
diff --git a/call/fake_network_pipe.cc b/call/fake_network_pipe.cc
index b5c0cb5..46adcb4 100644
--- a/call/fake_network_pipe.cc
+++ b/call/fake_network_pipe.cc
@@ -24,7 +24,6 @@
 namespace webrtc {
 
 namespace {
-constexpr int64_t kDefaultProcessIntervalMs = 5;
 constexpr int64_t kLogIntervalMs = 5000;
 }  // namespace
 
@@ -167,12 +166,6 @@
     packets_in_flight_.pop_back();
     ++dropped_packets_;
   }
-  if (network_behavior_->NextDeliveryTimeUs()) {
-    rtc::CritScope crit(&process_thread_lock_);
-    if (process_thread_)
-      process_thread_->WakeUp(nullptr);
-  }
-
   return sent;
 }
 
@@ -292,19 +285,14 @@
   }
 }
 
-int64_t FakeNetworkPipe::TimeUntilNextProcess() {
+absl::optional<int64_t> FakeNetworkPipe::TimeUntilNextProcess() {
   rtc::CritScope crit(&process_lock_);
   absl::optional<int64_t> delivery_us = network_behavior_->NextDeliveryTimeUs();
   if (delivery_us) {
     int64_t delay_us = *delivery_us - clock_->TimeInMicroseconds();
     return std::max<int64_t>((delay_us + 500) / 1000, 0);
   }
-  return kDefaultProcessIntervalMs;
-}
-
-void FakeNetworkPipe::ProcessThreadAttached(ProcessThread* process_thread) {
-  rtc::CritScope cs(&process_thread_lock_);
-  process_thread_ = process_thread;
+  return absl::nullopt;
 }
 
 bool FakeNetworkPipe::HasTransport() const {
diff --git a/call/fake_network_pipe.h b/call/fake_network_pipe.h
index 2c41dbf..661815b 100644
--- a/call/fake_network_pipe.h
+++ b/call/fake_network_pipe.h
@@ -86,8 +86,7 @@
 
 // Class faking a network link, internally is uses an implementation of a
 // SimulatedNetworkInterface to simulate network behavior.
-class FakeNetworkPipe : public webrtc::SimulatedPacketReceiverInterface,
-                        public Transport {
+class FakeNetworkPipe : public SimulatedPacketReceiverInterface {
  public:
   // Will keep |network_behavior| alive while pipe is alive itself.
   // Use these constructors if you plan to insert packets using DeliverPacket().
@@ -119,8 +118,8 @@
   // constructor.
   bool SendRtp(const uint8_t* packet,
                size_t length,
-               const PacketOptions& options) override;
-  bool SendRtcp(const uint8_t* packet, size_t length) override;
+               const PacketOptions& options);
+  bool SendRtcp(const uint8_t* packet, size_t length);
 
   // Implements the PacketReceiver interface. When/if packets are delivered,
   // they will be passed directly to the receiver instance given in
@@ -138,8 +137,7 @@
   // Processes the network queues and trigger PacketReceiver::IncomingPacket for
   // packets ready to be delivered.
   void Process() override;
-  int64_t TimeUntilNextProcess() override;
-  void ProcessThreadAttached(ProcessThread* process_thread) override;
+  absl::optional<int64_t> TimeUntilNextProcess() override;
 
   // Get statistics.
   float PercentageLoss();
@@ -193,10 +191,6 @@
   // |process_lock| guards the data structures involved in delay and loss
   // processes, such as the packet queues.
   rtc::CriticalSection process_lock_;
-
-  rtc::CriticalSection process_thread_lock_;
-  ProcessThread* process_thread_ RTC_GUARDED_BY(process_thread_lock_) = nullptr;
-
   // Packets  are added at the back of the deque, this makes the deque ordered
   // by increasing send time. The common case when removing packets from the
   // deque is removing early packets, which will be close to the front of the
diff --git a/call/simulated_network.cc b/call/simulated_network.cc
index 0884b29..c80255f 100644
--- a/call/simulated_network.cc
+++ b/call/simulated_network.cc
@@ -20,6 +20,9 @@
 #include "rtc_base/checks.h"
 
 namespace webrtc {
+namespace {
+constexpr int64_t kDefaultProcessDelayUs = 5000;
+}
 
 SimulatedNetwork::SimulatedNetwork(SimulatedNetwork::Config config,
                                    uint64_t random_seed)
@@ -76,15 +79,16 @@
   // calculated in UpdateCapacityQueue.
   queue_size_bytes_ += packet.size;
   capacity_link_.push({packet, packet.send_time_us});
+  if (!next_process_time_us_) {
+    next_process_time_us_ = packet.send_time_us + kDefaultProcessDelayUs;
+  }
 
   return true;
 }
 
 absl::optional<int64_t> SimulatedNetwork::NextDeliveryTimeUs() const {
   RTC_DCHECK_RUNS_SERIALIZED(&process_checker_);
-  if (!delay_link_.empty())
-    return delay_link_.begin()->arrival_time_us;
-  return absl::nullopt;
+  return next_process_time_us_;
 }
 
 void SimulatedNetwork::UpdateCapacityQueue(ConfigState state,
@@ -198,6 +202,14 @@
         PacketDeliveryInfo(packet_info.packet, packet_info.arrival_time_us));
     delay_link_.pop_front();
   }
+
+  if (!delay_link_.empty()) {
+    next_process_time_us_ = delay_link_.front().arrival_time_us;
+  } else if (!capacity_link_.empty()) {
+    next_process_time_us_ = receive_time_us + kDefaultProcessDelayUs;
+  } else {
+    next_process_time_us_.reset();
+  }
   return packets_to_deliver;
 }
 
diff --git a/call/simulated_network.h b/call/simulated_network.h
index 6adb412..632eb5d 100644
--- a/call/simulated_network.h
+++ b/call/simulated_network.h
@@ -87,6 +87,8 @@
   int64_t pending_drain_bits_ RTC_GUARDED_BY(process_checker_) = 0;
   absl::optional<int64_t> last_capacity_link_visit_us_
       RTC_GUARDED_BY(process_checker_);
+  absl::optional<int64_t> next_process_time_us_
+      RTC_GUARDED_BY(process_checker_);
 };
 
 }  // namespace webrtc
diff --git a/call/simulated_packet_receiver.h b/call/simulated_packet_receiver.h
index 03d7e96..2db46e8 100644
--- a/call/simulated_packet_receiver.h
+++ b/call/simulated_packet_receiver.h
@@ -13,13 +13,12 @@
 
 #include "api/test/simulated_network.h"
 #include "call/packet_receiver.h"
-#include "modules/include/module.h"
 
 namespace webrtc {
 
 // Private API that is fixing surface between DirectTransport and underlying
 // network conditions simulation implementation.
-class SimulatedPacketReceiverInterface : public PacketReceiver, public Module {
+class SimulatedPacketReceiverInterface : public PacketReceiver {
  public:
   // Must not be called in parallel with DeliverPacket or Process.
   // Destination receiver will be injected with this method
@@ -27,6 +26,15 @@
 
   // Reports average packet delay.
   virtual int AverageDelay() = 0;
+
+  // Process any pending tasks such as timeouts.
+  // Called on a worker thread.
+  virtual void Process() = 0;
+
+  // Returns the time until next process or nullopt to indicate that the next
+  // process time is unknown. If the next process time is unknown, this should
+  // be checked again any time a packet is enqueued.
+  virtual absl::optional<int64_t> TimeUntilNextProcess() = 0;
 };
 
 }  // namespace webrtc
diff --git a/call/test/fake_network_pipe_unittest.cc b/call/test/fake_network_pipe_unittest.cc
index 9f2a663..b8c7e56 100644
--- a/call/test/fake_network_pipe_unittest.cc
+++ b/call/test/fake_network_pipe_unittest.cc
@@ -259,7 +259,7 @@
 
   // Check that all the packets were sent.
   EXPECT_EQ(static_cast<size_t>(2 * kNumPackets), pipe->SentPackets());
-  fake_clock_.AdvanceTimeMilliseconds(pipe->TimeUntilNextProcess());
+  fake_clock_.AdvanceTimeMilliseconds(*pipe->TimeUntilNextProcess());
   EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
   pipe->Process();
 }
@@ -307,7 +307,7 @@
 
   // Check that all the packets were sent.
   EXPECT_EQ(static_cast<size_t>(kNumPackets), pipe->SentPackets());
-  fake_clock_.AdvanceTimeMilliseconds(pipe->TimeUntilNextProcess());
+  fake_clock_.AdvanceTimeMilliseconds(*pipe->TimeUntilNextProcess());
   EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
   pipe->Process();
 }
diff --git a/test/direct_transport.cc b/test/direct_transport.cc
index fd73691..b755449 100644
--- a/test/direct_transport.cc
+++ b/test/direct_transport.cc
@@ -50,19 +50,18 @@
 }
 
 DirectTransport::~DirectTransport() {
-  RTC_DCHECK_CALLED_SEQUENTIALLY(&sequence_checker_);
-  // Constructor updates |next_scheduled_task_|, so it's guaranteed to
-  // be initialized.
-  task_queue_->CancelTask(next_scheduled_task_);
+  if (next_process_task_)
+    task_queue_->CancelTask(*next_process_task_);
 }
 
 void DirectTransport::StopSending() {
-  RTC_DCHECK_CALLED_SEQUENTIALLY(&sequence_checker_);
-  task_queue_->CancelTask(next_scheduled_task_);
+  rtc::CritScope cs(&process_lock_);
+  if (next_process_task_)
+    task_queue_->CancelTask(*next_process_task_);
 }
 
 void DirectTransport::SetReceiver(PacketReceiver* receiver) {
-  RTC_DCHECK_CALLED_SEQUENTIALLY(&sequence_checker_);
+  rtc::CritScope cs(&process_lock_);
   fake_network_->SetReceiver(receiver);
 }
 
@@ -92,6 +91,9 @@
   int64_t send_time = clock_->TimeInMicroseconds();
   fake_network_->DeliverPacket(media_type, rtc::CopyOnWriteBuffer(data, length),
                                send_time);
+  rtc::CritScope cs(&process_lock_);
+  if (!next_process_task_)
+    ProcessPackets();
 }
 
 int DirectTransport::GetAverageDelayMs() {
@@ -104,17 +106,20 @@
     send_call_->SignalChannelNetworkState(MediaType::AUDIO, kNetworkUp);
     send_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
   }
-  SendPackets();
 }
 
-void DirectTransport::SendPackets() {
-  RTC_DCHECK_CALLED_SEQUENTIALLY(&sequence_checker_);
-
-  fake_network_->Process();
-
-  int64_t delay_ms = fake_network_->TimeUntilNextProcess();
-  next_scheduled_task_ =
-      task_queue_->PostDelayedTask([this]() { SendPackets(); }, delay_ms);
+void DirectTransport::ProcessPackets() {
+  next_process_task_.reset();
+  auto delay_ms = fake_network_->TimeUntilNextProcess();
+  if (delay_ms) {
+    next_process_task_ = task_queue_->PostDelayedTask(
+        [this]() {
+          fake_network_->Process();
+          rtc::CritScope cs(&process_lock_);
+          ProcessPackets();
+        },
+        *delay_ms);
+  }
 }
 }  // namespace test
 }  // namespace webrtc
diff --git a/test/direct_transport.h b/test/direct_transport.h
index f926ec5..d70748f 100644
--- a/test/direct_transport.h
+++ b/test/direct_transport.h
@@ -60,7 +60,7 @@
   int GetAverageDelayMs();
 
  private:
-  void SendPackets();
+  void ProcessPackets() RTC_EXCLUSIVE_LOCKS_REQUIRED(&process_lock_);
   void SendPacket(const uint8_t* data, size_t length);
   void Start();
 
@@ -68,13 +68,14 @@
   Clock* const clock_;
 
   SingleThreadedTaskQueueForTesting* const task_queue_;
-  SingleThreadedTaskQueueForTesting::TaskId next_scheduled_task_
-      RTC_GUARDED_BY(&sequence_checker_);
+
+  rtc::CriticalSection process_lock_;
+  absl::optional<SingleThreadedTaskQueueForTesting::TaskId> next_process_task_
+      RTC_GUARDED_BY(&process_lock_);
 
   const Demuxer demuxer_;
   const std::unique_ptr<SimulatedPacketReceiverInterface> fake_network_;
 
-  rtc::SequencedTaskChecker sequence_checker_;
 };
 }  // namespace test
 }  // namespace webrtc