// Copyright 2012 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "services/network/p2p/socket_udp.h"

#include <stdint.h>

#include <cmath>
#include <optional>
#include <utility>
#include <vector>

#include "base/check.h"
#include "base/containers/circular_deque.h"
#include "base/functional/bind.h"
#include "base/memory/raw_ptr.h"
#include "base/notimplemented.h"
#include "base/notreached.h"
#include "base/numerics/safe_conversions.h"
#include "base/run_loop.h"
#include "base/strings/string_number_conversions.h"
#include "base/test/metrics/histogram_tester.h"
#include "base/test/run_until.h"
#include "base/test/scoped_feature_list.h"
#include "base/test/task_environment.h"
#include "base/time/time.h"
#include "base/unguessable_token.h"
#include "mojo/public/cpp/bindings/pending_receiver.h"
#include "mojo/public/cpp/bindings/pending_remote.h"
#include "net/base/completion_once_callback.h"
#include "net/base/features.h"
#include "net/base/io_buffer.h"
#include "net/base/ip_endpoint.h"
#include "net/base/net_errors.h"
#include "net/base/network_anonymization_key.h"
#include "net/base/port_util.h"
#include "net/log/net_log_with_source.h"
#include "net/socket/datagram_server_socket.h"
#include "net/traffic_annotation/network_traffic_annotation_test_helper.h"
#include "services/network/p2p/socket_test_utils.h"
#include "services/network/p2p/socket_throttler.h"
#include "services/network/public/cpp/p2p_socket_type.h"
#include "services/network/throttling/network_conditions.h"
#include "services/network/throttling/throttling_controller.h"
#include "services/network/throttling/throttling_p2p_network_interceptor.h"
#include "testing/gmock/include/gmock/gmock.h"
#include "testing/gtest/include/gtest/gtest.h"
#include "third_party/webrtc/rtc_base/time_utils.h"

using ::testing::_;
using ::testing::DeleteArg;
using ::testing::DoAll;
using ::testing::ElementsAre;
using ::testing::Field;
using ::testing::InSequence;
using ::testing::Return;

namespace {

// TODO(nisse): We can't currently use webrtc::ScopedFakeClock, because
// we don't link with webrtc rtc_base_tests_utils. So roll our own.

// Creating an object of this class makes webrtc::TimeMicros() and
// related functions return zero unless the clock is advanced.
class ScopedFakeClock : public webrtc::ClockInterface {
 public:
  ScopedFakeClock() { prev_clock_ = webrtc::SetClockForTesting(this); }
  ~ScopedFakeClock() override { webrtc::SetClockForTesting(prev_clock_); }
  // ClockInterface implementation.
  int64_t TimeNanos() const override { return time_nanos_; }
  void SetTimeNanos(uint64_t time_nanos) { time_nanos_ = time_nanos; }

 private:
  raw_ptr<ClockInterface> prev_clock_;
  uint64_t time_nanos_ = 0;
};

class FakeDatagramServerSocket : public net::DatagramServerSocket {
 public:
  typedef std::
      tuple<net::IPEndPoint, std::vector<uint8_t>, std::optional<uint64_t>>
          UDPPacket;

  // P2PSocketUdp destroys a socket on errors so sent packets
  // need to be stored outside of this object.
  FakeDatagramServerSocket(base::circular_deque<UDPPacket>* sent_packets,
                           std::vector<uint16_t>* used_ports,
                           ScopedFakeClock* fake_clock)
      : sent_packets_(sent_packets),
        recv_address_(nullptr),
        recv_size_(0),
        used_ports_(used_ports),
        fake_clock_ptr_(fake_clock) {}

  void Close() override {}

  int GetPeerAddress(net::IPEndPoint* address) const override { NOTREACHED(); }

  int GetLocalAddress(net::IPEndPoint* address) const override {
    *address = address_;
    return 0;
  }

  void UseNonBlockingIO() override {}

  int Listen(const net::IPEndPoint& address) override {
    if (used_ports_) {
      for (auto used_port : *used_ports_) {
        if (used_port == address.port())
          return -1;
      }
      used_ports_->push_back(address.port());
    }

    address_ = address;
    return 0;
  }

  int RecvFrom(net::IOBuffer* buf,
               int buf_len,
               net::IPEndPoint* address,
               net::CompletionOnceCallback callback) override {
    CHECK(recv_callback_.is_null());

    if (read_error_result_) {
      // Return the simulated error.
      return *read_error_result_;
    }

    if (incoming_packets_.size() > 0) {
      scoped_refptr<net::IOBuffer> buffer(buf);
      const UDPPacket& front_packet = incoming_packets_.front();
      const std::vector<uint8_t>& front_packet_data = std::get<1>(front_packet);

      size_t size = std::min(front_packet_data.size(),
                             base::checked_cast<size_t>(buf_len));
      buffer->span().copy_prefix_from(
          base::span(front_packet_data).first(size));
      *address = std::get<0>(front_packet);
      std::optional<uint64_t> received_time = std::get<2>(front_packet);
      if (received_time) {
        fake_clock_ptr_->SetTimeNanos(*received_time);
      }
      incoming_packets_.pop_front();
      return size;
    } else {
      recv_callback_ = std::move(callback);
      recv_buffer_ = buf;
      recv_size_ = buf_len;
      recv_address_ = address;
      return net::ERR_IO_PENDING;
    }
  }

  int DoSend(bool is_async_send,
             int send_result,
             scoped_refptr<net::IOBuffer> buf,
             int buf_len,
             const net::IPEndPoint& address,
             net::CompletionOnceCallback callback) {
    if (send_result != net::OK) {
      // Return `send_result` to simulate failure.
      if (is_async_send) {
        std::move(callback).Run(send_result);
      }
      return send_result;
    }

    base::span<const uint8_t> to_write =
        buf->first(base::checked_cast<size_t>(buf_len));
    std::vector<uint8_t> data_vector(to_write.begin(), to_write.end());
    sent_packets_->push_back(
        UDPPacket(address, std::move(data_vector), std::nullopt));

    if (is_async_send) {
      std::move(callback).Run(buf_len);
    }
    return buf_len;
  }

  int SendTo(net::IOBuffer* buf,
             int buf_len,
             const net::IPEndPoint& address,
             net::CompletionOnceCallback callback) override {
    int send_result = PopNextSendResult();
    bool is_async_send = send_result == net::ERR_IO_PENDING;
    if (is_async_send) {
      scoped_refptr<net::IOBuffer> buffer(buf);
      send_result = PopNextSendResult();

      base::SequencedTaskRunner::GetCurrentDefault()->PostTask(
          FROM_HERE,
          base::BindOnce(base::IgnoreResult(&FakeDatagramServerSocket::DoSend),
                         base::Unretained(this), /*is_async_send=*/true,
                         send_result, std::move(buffer), buf_len, address,
                         std::move(callback)));
      return net::ERR_IO_PENDING;
    }
    return DoSend(/*is_async_send=*/false, send_result, buf, buf_len, address,
                  std::move(callback));
  }

  base::circular_deque<int>& send_result_queue() { return send_result_queue_; }

  // Returns and pops the front of `send_result_queue_`.  Returns `net::OK` when
  // `send_result_queue_` is empty.
  int PopNextSendResult() {
    if (send_result_queue_.empty()) {
      return net::OK;
    }

    int next_send_result = send_result_queue_.front();
    send_result_queue_.pop_front();
    return next_send_result;
  }

  void SetReadErrorResult(std::optional<int> value) {
    read_error_result_ = value;
  }

  int SetReceiveBufferSize(int32_t size) override { return net::OK; }

  int SetSendBufferSize(int32_t size) override { return net::OK; }

  int SetDoNotFragment() override { return net::OK; }

  int SetRecvTos() override {
    is_recv_ecn_enabled_ = true;
    return net::OK;
  }

  void SetMsgConfirm(bool confirm) override {}

  void ReceivePacket(const net::IPEndPoint& address,
                     std::vector<uint8_t> data) {
    AddRecvPacket(address, data);
    FireRecvCallback();
  }

  // Add a packet into the buffer, and specify the fake clock time when
  // the packet is received by socket.
  void AddRecvPacket(
      const net::IPEndPoint& address,
      const std::vector<uint8_t> data,
      const std::optional<uint64_t> received_time = std::nullopt) {
    incoming_packets_.push_back(UDPPacket(address, data, received_time));
  }

  void FireRecvCallback() {
    if (!recv_callback_.is_null()) {
      DCHECK(!incoming_packets_.empty());
      const UDPPacket& front_packet = incoming_packets_.front();
      const auto& front_packet_data = std::get<1>(front_packet);
      size_t size = std::min(base::checked_cast<size_t>(recv_size_),
                             front_packet_data.size());
      recv_buffer_->span().copy_prefix_from(
          base::span(front_packet_data).first(size));
      *recv_address_ = std::get<0>(front_packet);
      std::optional<uint64_t> received_time = std::get<2>(front_packet);
      if (received_time) {
        fake_clock_ptr_->SetTimeNanos(*received_time);
      }
      incoming_packets_.pop_front();
      recv_buffer_ = nullptr;
      std::move(recv_callback_).Run(size);
    }
  }

  const net::NetLogWithSource& NetLog() const override { return net_log_; }

  void AllowAddressReuse() override { NOTIMPLEMENTED(); }

  void AllowBroadcast() override { NOTIMPLEMENTED(); }

  void AllowAddressSharingForMulticast() override { NOTIMPLEMENTED(); }

  int JoinGroup(const net::IPAddress& group_address) const override {
    NOTIMPLEMENTED();
    return net::ERR_NOT_IMPLEMENTED;
  }

  int LeaveGroup(const net::IPAddress& group_address) const override {
    NOTIMPLEMENTED();
    return net::ERR_NOT_IMPLEMENTED;
  }

  int SetMulticastInterface(uint32_t interface_index) override {
    NOTIMPLEMENTED();
    return net::ERR_NOT_IMPLEMENTED;
  }

  int SetMulticastTimeToLive(int time_to_live) override {
    NOTIMPLEMENTED();
    return net::ERR_NOT_IMPLEMENTED;
  }

  int SetMulticastLoopbackMode(bool loopback) override {
    NOTIMPLEMENTED();
    return net::ERR_NOT_IMPLEMENTED;
  }

  int SetDiffServCodePoint(net::DiffServCodePoint dscp) override {
    NOTIMPLEMENTED();
    return net::ERR_NOT_IMPLEMENTED;
  }

  int SetTos(net::DiffServCodePoint dscp, net::EcnCodePoint ecn) override {
    set_tos_call_count_++;
    if (set_tos_result_ != net::OK) {
      return set_tos_result_;
    }

    if (dscp != net::DSCP_NO_CHANGE) {
      last_sent_dscp_ = dscp;
    }
    if (ecn != net::ECN_NO_CHANGE) {
      last_sent_ecn_ = ecn;
    }
    return net::OK;
  }

  void DetachFromThread() override { NOTIMPLEMENTED(); }

  net::DscpAndEcn GetLastTos() const override {
    if (!is_recv_ecn_enabled_) {
      return {net::DSCP_DEFAULT, net::ECN_DEFAULT};
    } else {
      return {net::DSCP_DEFAULT, net::ECN_ECT1};
    }
  }

  net::EcnCodePoint GetLastSentEcn() const { return last_sent_ecn_; }
  net::DiffServCodePoint GetLastSentDscp() const { return last_sent_dscp_; }

  void SetSetTosResult(int result) { set_tos_result_ = result; }
  int set_tos_call_count() const { return set_tos_call_count_; }

 private:
  int set_tos_result_ = net::OK;
  int set_tos_call_count_ = 0;
  bool is_recv_ecn_enabled_ = false;
  net::DiffServCodePoint last_sent_dscp_ = net::DSCP_DEFAULT;
  net::EcnCodePoint last_sent_ecn_ = net::ECN_DEFAULT;
  net::IPEndPoint address_;
  raw_ptr<base::circular_deque<UDPPacket>> sent_packets_;
  base::circular_deque<UDPPacket> incoming_packets_;
  net::NetLogWithSource net_log_;

  scoped_refptr<net::IOBuffer> recv_buffer_;
  raw_ptr<net::IPEndPoint> recv_address_;
  int recv_size_;
  net::CompletionOnceCallback recv_callback_;
  raw_ptr<std::vector<uint16_t>> used_ports_;

  // Tests may push error codes to simulate `SendTo()` failures.  Tests assume
  // success by default when `send_result_queue_` is empty.  Push
  // `net::ERR_IO_PENDING` to simulate an async send.
  base::circular_deque<int> send_result_queue_;

  // Tests may set this to simulate read errors.
  std::optional<int> read_error_result_;

  // Owned by |P2PSocketUdpTest|.
  raw_ptr<ScopedFakeClock> fake_clock_ptr_;
};

std::unique_ptr<net::DatagramServerSocket> CreateFakeDatagramServerSocket(
    base::circular_deque<FakeDatagramServerSocket::UDPPacket>* sent_packets,
    std::vector<uint16_t>* used_ports,
    ScopedFakeClock* fake_clock,
    net::NetLog* net_log) {
  return std::make_unique<FakeDatagramServerSocket>(sent_packets, used_ports,
                                                    fake_clock);
}

}  // namespace

namespace network {

class P2PSocketUdpTest : public testing::Test {
 public:
  P2PSocketUdpTest() : P2PSocketUdpTest(std::nullopt) {}

 protected:
  explicit P2PSocketUdpTest(
      std::optional<base::UnguessableToken> devtools_token)
      : devtools_token_(devtools_token),
        net_log_with_source_(
            net::NetLogWithSource::Make(net::NetLog::Get(),
                                        net::NetLogSourceType::UDP_SOCKET)) {}

  // It is the helper method to get easy access to matcher.
  MOCK_METHOD(void,
              SinglePacketReceptionHelper,
              (const net::IPEndPoint& socket_address,
               base::span<const uint8_t> data,
               base::TimeTicks timestamp));

  void SetUp() override {
    mojo::PendingRemote<mojom::P2PSocketClient> socket_client;
    mojo::PendingRemote<mojom::P2PSocket> socket;
    auto socket_receiver = socket.InitWithNewPipeAndPassReceiver();

    fake_client_ = std::make_unique<FakeSocketClient>(
        std::move(socket), socket_client.InitWithNewPipeAndPassReceiver());

    EXPECT_CALL(*fake_client_.get(), SocketCreated(_, _)).Times(1);

    // Unpack received batching packets for testing.
    ON_CALL(*fake_client_.get(), DataReceived(_))
        .WillByDefault(
            [this](std::vector<network::mojom::P2PReceivedPacketPtr> packets) {
              for (auto& packet : packets) {
                SinglePacketReceptionHelper(packet->socket_address,
                                            packet->data, packet->timestamp);
                received_packets_.emplace_back(std::move(packet));
              }
              return;
            });

    socket_impl_ = std::make_unique<P2PSocketUdp>(
        &socket_delegate_, std::move(socket_client), std::move(socket_receiver),
        &throttler_, TRAFFIC_ANNOTATION_FOR_TESTS,
        net_log_with_source_.net_log(),
        base::BindRepeating(&CreateFakeDatagramServerSocket, &sent_packets_,
                            nullptr, &fake_clock_),
        devtools_token_);

    local_address_ = ParseAddress(kTestLocalIpAddress, kTestPort1);
    socket_impl_->Init(
        local_address_, 0, 0,
        P2PHostAndIPEndPoint(std::string(),
                             ParseAddress(kTestIpAddress1, kTestPort1)),
        net::NetworkAnonymizationKey());
    socket_ = GetSocketFromHost(socket_impl_.get());

    dest1_ = ParseAddress(kTestIpAddress1, kTestPort1);
    dest2_ = ParseAddress(kTestIpAddress2, kTestPort2);
  }

  static FakeDatagramServerSocket* GetSocketFromHost(
      P2PSocketUdp* socket_host) {
    return static_cast<FakeDatagramServerSocket*>(socket_host->socket_.get());
  }

  base::test::TaskEnvironment task_environment_{
      base::test::TaskEnvironment::TimeSource::MOCK_TIME};
  P2PMessageThrottler throttler_;
  ScopedFakeClock fake_clock_;
  base::circular_deque<FakeDatagramServerSocket::UDPPacket> sent_packets_;
  base::circular_deque<mojo::StructPtr<network::mojom::P2PReceivedPacket>>
      received_packets_;
  FakeP2PSocketDelegate socket_delegate_;
  std::unique_ptr<P2PSocketUdp> socket_impl_;
  raw_ptr<FakeDatagramServerSocket> socket_;  // Owned by |socket_impl_|.
  std::unique_ptr<FakeSocketClient> fake_client_;
  std::optional<base::UnguessableToken> devtools_token_ = std::nullopt;
  net::NetLogWithSource net_log_with_source_;

  net::IPEndPoint local_address_;

  net::IPEndPoint dest1_;
  net::IPEndPoint dest2_;
};

// Verify that we can send STUN messages before we receive anything
// from the other side.
TEST_F(P2PSocketUdpTest, SendStunNoAuth) {
  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(3);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet1;
  CreateStunRequest(&packet1);
  socket_impl_->Send(packet1, P2PPacketInfo(dest1_, options, 0));

  std::vector<uint8_t> packet2;
  CreateStunResponse(&packet2);
  socket_impl_->Send(packet2, P2PPacketInfo(dest1_, options, 0));

  std::vector<uint8_t> packet3;
  CreateStunError(&packet3);
  socket_impl_->Send(packet3, P2PPacketInfo(dest1_, options, 0));

  ASSERT_EQ(sent_packets_.size(), 3U);
  ASSERT_EQ(std::get<1>(sent_packets_[0]), packet1);
  ASSERT_EQ(std::get<1>(sent_packets_[1]), packet2);
  ASSERT_EQ(std::get<1>(sent_packets_[2]), packet3);

  base::RunLoop().RunUntilIdle();
}

// Verify that no data packets can be sent before STUN binding has
// finished.
TEST_F(P2PSocketUdpTest, SendDataNoAuth) {
  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  socket_ = nullptr;  // Since about to give up `socket_impl_`.
  auto* socket_impl_ptr = socket_impl_.get();
  socket_delegate_.ExpectDestruction(std::move(socket_impl_));
  socket_impl_ptr->Send(packet, P2PPacketInfo(dest1_, options, 0));

  ASSERT_EQ(sent_packets_.size(), 0U);

  base::RunLoop().RunUntilIdle();

  EXPECT_TRUE(fake_client_->connection_error());
}

TEST_F(P2PSocketUdpTest, SendRestrictedAddress) {
  base::test::ScopedFeatureList feature_list;
  int restricted_port = 12345;
  net::IPEndPoint restricted_dest = ParseAddress("127.0.0.1", restricted_port);
  feature_list.InitAndEnableFeatureWithParameters(
      net::features::kRestrictAbusePortsOnLocalhost,
      {{"localhost_restrict_ports", base::NumberToString(restricted_port)}});
  net::ReloadLocalhostRestrictedPortsForTesting();
  base::circular_deque<FakeDatagramServerSocket::UDPPacket> sent_packets;
  std::vector<uint16_t> used_ports;
  P2PSocketUdp::DatagramServerSocketFactory fake_socket_factory =
      base::BindRepeating(&CreateFakeDatagramServerSocket, &sent_packets,
                          &used_ports, &fake_clock_);
  P2PMessageThrottler throttler;

  mojo::PendingRemote<mojom::P2PSocketClient> socket_client;
  mojo::PendingRemote<mojom::P2PSocket> socket;
  auto socket_receiver = socket.InitWithNewPipeAndPassReceiver();

  FakeSocketClient fake_client2(std::move(socket),
                                socket_client.InitWithNewPipeAndPassReceiver());

  auto socket_impl = std::make_unique<P2PSocketUdp>(
      &socket_delegate_, std::move(socket_client), std::move(socket_receiver),
      &throttler, TRAFFIC_ANNOTATION_FOR_TESTS, /*net_log=*/nullptr,
      std::move(fake_socket_factory), std::nullopt);
  net::IPEndPoint local_address = ParseAddress(kTestLocalIpAddress, kTestPort1);

  auto* socket_impl_ptr = socket_impl.get();
  socket_delegate_.ExpectDestruction(std::move(socket_impl));
  socket_impl_ptr->Init(local_address, 0, 0,
                        P2PHostAndIPEndPoint(std::string(), restricted_dest),
                        net::NetworkAnonymizationKey());

  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  webrtc::AsyncSocketPacketOptions options;
  socket_impl_ptr->Send(request_packet,
                        P2PPacketInfo(restricted_dest, options, 0));

  ASSERT_TRUE(
      base::test::RunUntil([&]() { return fake_client2.connection_error(); }));
}

// Verify that we can send data after we've received STUN request
// from the other side.
TEST_F(P2PSocketUdpTest, SendAfterStunRequest) {
  // Receive packet from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  // Now we should be able to send any data to |dest1_|.
  EXPECT_CALL(*fake_client_.get(), SendComplete(_));

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));

  ASSERT_EQ(1U, sent_packets_.size());
  ASSERT_EQ(dest1_, std::get<0>(sent_packets_[0]));

  base::RunLoop().RunUntilIdle();
}

// Verify that we can send data after we've received STUN response
// from the other side.
TEST_F(P2PSocketUdpTest, SendAfterStunResponse) {
  // Receive packet from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunResponse(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  // Now we should be able to send any data to |dest1_|.
  EXPECT_CALL(*fake_client_.get(), SendComplete(_));

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));

  ASSERT_EQ(1U, sent_packets_.size());
  ASSERT_EQ(dest1_, std::get<0>(sent_packets_[0]));

  base::RunLoop().RunUntilIdle();
}

// Verify messages still cannot be sent to an unathorized host after
// successful binding with different host.
TEST_F(P2PSocketUdpTest, SendAfterStunResponseDifferentHost) {
  // Receive packet from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunResponse(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  // Should fail when trying to send the same packet to |dest2_|.
  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  socket_ = nullptr;  // Since about to give up `socket_impl_`.
  auto* socket_impl_ptr = socket_impl_.get();
  socket_delegate_.ExpectDestruction(std::move(socket_impl_));
  socket_impl_ptr->Send(packet, P2PPacketInfo(dest2_, options, 0));

  base::RunLoop().RunUntilIdle();

  EXPECT_TRUE(fake_client_->connection_error());
}

TEST_F(P2PSocketUdpTest, AsyncSend) {
  base::HistogramTester histograms;

  // Setup two successful async send operations.
  socket_->send_result_queue().push_back(net::ERR_IO_PENDING);
  socket_->send_result_queue().push_back(net::OK);
  socket_->send_result_queue().push_back(net::ERR_IO_PENDING);

  // Authorize sends to `dest1_` via a STUN request.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(2);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet1;
  CreateRandomPacket(&packet1);
  std::vector<uint8_t> packet2;
  CreateRandomPacket(&packet2);

  // First send returns `net::ERR_IO_PENDING`.
  socket_impl_->Send(packet1, P2PPacketInfo(dest1_, options, 0));
  ASSERT_EQ(0U, sent_packets_.size());

  // Second send must be queued because the first is still pending.
  socket_impl_->Send(packet2, P2PPacketInfo(dest1_, options, 1));
  ASSERT_EQ(0U, sent_packets_.size());

  // Wait for the async sends to complete.
  base::RunLoop().RunUntilIdle();

  ASSERT_EQ(2U, sent_packets_.size());
  EXPECT_EQ(dest1_, std::get<0>(sent_packets_[0]));
  EXPECT_EQ(packet1, std::get<1>(sent_packets_[0]));
  EXPECT_EQ(dest1_, std::get<0>(sent_packets_[1]));
  EXPECT_EQ(packet2, std::get<1>(sent_packets_[1]));
  histograms.ExpectUniqueSample("WebRTC.P2P.UDP.SendResult", net::OK, 2);
}

TEST_F(P2PSocketUdpTest, AsyncSendError) {
  base::HistogramTester histograms;

  // Setup an async send operation that fails.
  socket_->send_result_queue().push_back(net::ERR_IO_PENDING);
  socket_->send_result_queue().push_back(net::ERR_FAILED);

  // Authorize sends to `dest1_` via a STUN request.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  // The send returns `net::ERR_IO_PENDING` but the async task will invoke the
  // error callback, which must destroy the socket.
  socket_ = nullptr;
  auto* socket_impl_ptr = socket_impl_.get();
  socket_delegate_.ExpectDestruction(std::move(socket_impl_));
  socket_impl_ptr->Send(packet, P2PPacketInfo(dest1_, options, 0));

  // No packets should have been sent synchronously.
  ASSERT_EQ(0U, sent_packets_.size());

  // Run the posted async task which delivers the error.
  base::RunLoop().RunUntilIdle();

  // The socket should have been destroyed due to the error.
  EXPECT_TRUE(fake_client_->connection_error());
  ASSERT_EQ(0U, sent_packets_.size());
  histograms.ExpectUniqueSample("WebRTC.P2P.UDP.SendResult", net::ERR_FAILED,
                                1);
}

TEST_F(P2PSocketUdpTest, AsyncSendThenSyncSend) {
  // Set up a successful async send operation.
  socket_->send_result_queue().push_back(net::ERR_IO_PENDING);

  // Authorize sends to `dest1_` via a STUN request.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(0);

  // The async send callback, `P2PSocketUdp::OnSend()`, sends the second packet
  // and then runs both send completion callbacks.
  EXPECT_CALL(*fake_client_.get(), SendBatchComplete(_)).Times(1);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet1;
  CreateRandomPacket(&packet1);
  std::vector<uint8_t> packet2;
  CreateRandomPacket(&packet2);

  // First send returns `net::ERR_IO_PENDING`.
  socket_impl_->Send(packet1, P2PPacketInfo(dest1_, options, 0));
  ASSERT_EQ(0U, sent_packets_.size());

  // Second send must be queued because the first is still pending.
  socket_impl_->Send(packet2, P2PPacketInfo(dest1_, options, 1));
  ASSERT_EQ(0U, sent_packets_.size());

  // Wait for the async sends to complete.
  base::RunLoop().RunUntilIdle();

  ASSERT_EQ(2U, sent_packets_.size());
  EXPECT_EQ(dest1_, std::get<0>(sent_packets_[0]));
  EXPECT_EQ(packet1, std::get<1>(sent_packets_[0]));
  EXPECT_EQ(dest1_, std::get<0>(sent_packets_[1]));
  EXPECT_EQ(packet2, std::get<1>(sent_packets_[1]));
}

TEST_F(P2PSocketUdpTest, AsyncSendThenSyncError) {
  // Setup a successful async send followed by a sync send that fails.
  socket_->send_result_queue().push_back(net::ERR_IO_PENDING);
  socket_->send_result_queue().push_back(net::OK);
  socket_->send_result_queue().push_back(net::ERR_FAILED);

  // Authorize sends to `dest1_` via a STUN request.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  // The async send callback, `P2PSocketUdp::OnSend()`, fails to send the second
  // packet, which then errors the connection before running the first send
  // completion callback.
  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(0);
  EXPECT_CALL(*fake_client_.get(), SendBatchComplete(_)).Times(0);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet1;
  CreateRandomPacket(&packet1);
  std::vector<uint8_t> packet2;
  CreateRandomPacket(&packet2);

  // First send returns `net::ERR_IO_PENDING`.
  socket_impl_->Send(packet1, P2PPacketInfo(dest1_, options, 0));
  ASSERT_EQ(0U, sent_packets_.size());

  // Second send must be queued because the first is still pending.
  socket_ = nullptr;
  auto* socket_impl_ptr = socket_impl_.get();
  socket_delegate_.ExpectDestruction(std::move(socket_impl_));
  socket_impl_ptr->Send(packet2, P2PPacketInfo(dest1_, options, 0));

  // Wait for the async sends to complete.
  base::RunLoop().RunUntilIdle();

  // The first packet must send successfully.
  ASSERT_EQ(1U, sent_packets_.size());
  EXPECT_EQ(dest1_, std::get<0>(sent_packets_[0]));
  EXPECT_EQ(packet1, std::get<1>(sent_packets_[0]));

  // The second packet must fail.
  EXPECT_TRUE(fake_client_->connection_error());
}

TEST_F(P2PSocketUdpTest, SendPacketWithCustomDscpOrEcn) {
  // Open for sends to `dest1_`.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  socket_->ReceivePacket(dest1_, request_packet);

  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  // We'll send four packets.
  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(4);

  // Send with defaults.
  webrtc::AsyncSocketPacketOptions pkt1_options;
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, pkt1_options, 0));
  ASSERT_EQ(1U, sent_packets_.size());
  EXPECT_EQ(net::ECN_DEFAULT, socket_->GetLastSentEcn());
  EXPECT_EQ(net::DSCP_DEFAULT, socket_->GetLastSentDscp());

  // Send with ECT(1).
  webrtc::AsyncSocketPacketOptions pkt2_options;
  pkt2_options.ect_1 = true;
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, pkt2_options, 0));
  ASSERT_EQ(2U, sent_packets_.size());
  EXPECT_EQ(net::ECN_ECT1, socket_->GetLastSentEcn());
  EXPECT_EQ(net::DSCP_DEFAULT, socket_->GetLastSentDscp());

  // Send with DSCP::CS1 and ECT(1).
  webrtc::AsyncSocketPacketOptions pkt3_options;
  pkt3_options.ect_1 = true;
  pkt3_options.dscp = webrtc::DSCP_CS1;
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, pkt3_options, 0));
  ASSERT_EQ(3U, sent_packets_.size());
  EXPECT_EQ(net::ECN_ECT1, socket_->GetLastSentEcn());
  EXPECT_EQ(net::DSCP_CS1, socket_->GetLastSentDscp());

  // Send with only DSCP::CS4.
  webrtc::AsyncSocketPacketOptions pkt4_options;
  pkt4_options.dscp = webrtc::DSCP_CS4;
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, pkt4_options, 0));
  ASSERT_EQ(4U, sent_packets_.size());
  EXPECT_EQ(net::ECN_NOT_ECT, socket_->GetLastSentEcn());
  EXPECT_EQ(net::DSCP_CS4, socket_->GetLastSentDscp());

  base::RunLoop().RunUntilIdle();
}

TEST_F(P2PSocketUdpTest, SetTosBackoff) {
  // Open for sends to `dest1_`.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  socket_->ReceivePacket(dest1_, request_packet);

  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(4);

  // Fail SetTos and check that we back off.
  socket_->SetSetTosResult(net::ERR_FAILED);
  webrtc::AsyncSocketPacketOptions pkt1_options;
  pkt1_options.dscp = webrtc::DSCP_CS1;
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, pkt1_options, 0));
  EXPECT_EQ(1, socket_->set_tos_call_count());

  // The second send should not trigger a SetTos call.
  webrtc::AsyncSocketPacketOptions pkt2_options;
  pkt2_options.dscp = webrtc::DSCP_CS2;
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, pkt2_options, 0));
  EXPECT_EQ(1, socket_->set_tos_call_count());

  // After the backoff timeout we should try setting TOS again.
  // The backoff policy has an initial delay of 100ms.
  task_environment_.FastForwardBy(base::Milliseconds(100));
  // Now, make SetTos succeed and check that we don't back off anymore.
  socket_->SetSetTosResult(net::OK);
  webrtc::AsyncSocketPacketOptions pkt4_options;
  pkt4_options.dscp = webrtc::DSCP_CS4;
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, pkt4_options, 0));
  EXPECT_EQ(2, socket_->set_tos_call_count());

  // The backoff has been reset, so this should trigger a SetTos call.
  webrtc::AsyncSocketPacketOptions pkt5_options;
  pkt5_options.dscp = webrtc::DSCP_CS5;
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, pkt5_options, 0));
  EXPECT_EQ(3, socket_->set_tos_call_count());

  base::RunLoop().RunUntilIdle();
}

TEST_F(P2PSocketUdpTest, BatchesSendAfterSendingAllowed) {
  // Open for sends to `dest1_`.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  socket_->ReceivePacket(dest1_, request_packet);

  network::P2PPacketInfo info;
  info.destination = dest1_;
  std::vector<network::mojom::P2PSendPacketPtr> batch;
  info.packet_id = 1;
  std::vector<uint8_t> packet1;
  CreateRandomPacket(&packet1);
  batch.push_back(network::mojom::P2PSendPacket::New(packet1, info));
  info.packet_id = 2;
  std::vector<uint8_t> packet2;
  CreateRandomPacket(&packet2);
  batch.push_back(network::mojom::P2PSendPacket::New(packet2, info));
  socket_impl_->SendBatch(std::move(batch));
  ASSERT_EQ(sent_packets_.size(), 2u);
  EXPECT_CALL(*fake_client_, SendBatchComplete(ElementsAre(
                                 Field(&P2PSendPacketMetrics::packet_id, 1),
                                 Field(&P2PSendPacketMetrics::packet_id, 2))));
  base::RunLoop().RunUntilIdle();
}

// Verify throttler not allowing unlimited sending of ICE messages to
// any destination.
TEST_F(P2PSocketUdpTest, ThrottleAfterLimit) {
  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(3);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet1;
  CreateStunRequest(&packet1);
  throttler_.SetSendIceBandwidth(packet1.size() * 2);
  socket_impl_->Send(packet1, P2PPacketInfo(dest1_, options, 0));
  socket_impl_->Send(packet1, P2PPacketInfo(dest2_, options, 0));

  net::IPEndPoint dest3 = ParseAddress(kTestIpAddress1, 2222);
  // This packet must be dropped by the throttler.
  socket_impl_->Send(packet1, P2PPacketInfo(dest3, options, 0));
  ASSERT_EQ(sent_packets_.size(), 2U);

  base::RunLoop().RunUntilIdle();
}

// Verify we can send packets to a known destination when ICE throttling is
// active.
TEST_F(P2PSocketUdpTest, ThrottleAfterLimitAfterReceive) {
  // Receive packet from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _))
      .Times(1);
  socket_->ReceivePacket(dest1_, request_packet);

  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(6);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet1;
  CreateStunRequest(&packet1);
  throttler_.SetSendIceBandwidth(packet1.size());
  // |dest1_| is known address, throttling will not be applied.
  socket_impl_->Send(packet1, P2PPacketInfo(dest1_, options, 0));
  // Trying to send the packet to dest1_ in the same window. It should go.
  socket_impl_->Send(packet1, P2PPacketInfo(dest1_, options, 0));

  // Throttler should allow this packet to go through.
  socket_impl_->Send(packet1, P2PPacketInfo(dest2_, options, 0));

  net::IPEndPoint dest3 = ParseAddress(kTestIpAddress1, 2223);
  // This packet will be dropped, as limit only for a single packet.
  socket_impl_->Send(packet1, P2PPacketInfo(dest3, options, 0));
  net::IPEndPoint dest4 = ParseAddress(kTestIpAddress1, 2224);
  // This packet should also be dropped.
  socket_impl_->Send(packet1, P2PPacketInfo(dest4, options, 0));
  // |dest1| is known, we can send as many packets to it.
  socket_impl_->Send(packet1, P2PPacketInfo(dest1_, options, 0));
  ASSERT_EQ(sent_packets_.size(), 4U);

  base::RunLoop().RunUntilIdle();
}

// Test that once the limit is hit, the throttling stops at the expected time,
// allowing packets to be sent again.
TEST_F(P2PSocketUdpTest, ThrottlingStopsAtExpectedTimes) {
  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(12);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateStunRequest(&packet);
  // Limit of 2 packets per second.
  throttler_.SetSendIceBandwidth(packet.size() * 2);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));
  socket_impl_->Send(packet, P2PPacketInfo(dest2_, options, 0));
  EXPECT_EQ(2U, sent_packets_.size());

  // These packets must be dropped by the throttler since the limit was hit and
  // the time hasn't advanced.
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));
  socket_impl_->Send(packet, P2PPacketInfo(dest2_, options, 0));
  EXPECT_EQ(2U, sent_packets_.size());

  // Advance the time to 0.999 seconds; throttling should still just barely be
  // active.
  fake_clock_.SetTimeNanos(webrtc::kNumNanosecsPerMillisec * 999);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));
  socket_impl_->Send(packet, P2PPacketInfo(dest2_, options, 0));
  EXPECT_EQ(2U, sent_packets_.size());

  // After hitting the second mark, we should be able to send again.
  // Add an extra millisecond to account for rounding errors.
  fake_clock_.SetTimeNanos(webrtc::kNumNanosecsPerMillisec * 1001);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));
  EXPECT_EQ(3U, sent_packets_.size());

  // This time, hit the limit in the middle of the period.
  fake_clock_.SetTimeNanos(webrtc::kNumNanosecsPerMillisec * 1500);
  socket_impl_->Send(packet, P2PPacketInfo(dest2_, options, 0));
  EXPECT_EQ(4U, sent_packets_.size());

  // Again, throttling should be active until the next second mark.
  fake_clock_.SetTimeNanos(webrtc::kNumNanosecsPerMillisec * 1999);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));
  socket_impl_->Send(packet, P2PPacketInfo(dest2_, options, 0));
  EXPECT_EQ(4U, sent_packets_.size());
  fake_clock_.SetTimeNanos(webrtc::kNumNanosecsPerMillisec * 2002);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));
  socket_impl_->Send(packet, P2PPacketInfo(dest2_, options, 0));
  EXPECT_EQ(6U, sent_packets_.size());

  base::RunLoop().RunUntilIdle();
}

// Verify that we can open UDP sockets listening in a given port range,
// and fail if all ports in the range are already in use.
TEST_F(P2PSocketUdpTest, PortRangeImplicitPort) {
  const uint16_t min_port = 10000;
  const uint16_t max_port = 10001;
  base::circular_deque<FakeDatagramServerSocket::UDPPacket> sent_packets;
  std::vector<uint16_t> used_ports;
  P2PSocketUdp::DatagramServerSocketFactory fake_socket_factory =
      base::BindRepeating(&CreateFakeDatagramServerSocket, &sent_packets,
                          &used_ports, &fake_clock_);
  P2PMessageThrottler throttler;

  mojo::PendingRemote<mojom::P2PSocketClient> socket_client;
  auto socket_client_receiver = socket_client.InitWithNewPipeAndPassReceiver();
  mojo::PendingRemote<mojom::P2PSocket> socket;
  auto socket_receiver = socket.InitWithNewPipeAndPassReceiver();

  FakeSocketClient fake_client2(std::move(socket),
                                std::move(socket_client_receiver));
  EXPECT_CALL(fake_client2, SocketCreated(_, _)).Times(max_port - min_port + 1);

  for (unsigned port = min_port; port <= max_port; ++port) {
    std::unique_ptr<P2PSocketUdp> socket_impl(new P2PSocketUdp(
        &socket_delegate_, std::move(socket_client), std::move(socket_receiver),
        &throttler, TRAFFIC_ANNOTATION_FOR_TESTS, /*net_log=*/nullptr,
        fake_socket_factory, std::nullopt));
    net::IPEndPoint local_address = ParseAddress(kTestLocalIpAddress, 0);
    socket_impl->Init(
        local_address, min_port, max_port,
        P2PHostAndIPEndPoint(std::string(),
                             ParseAddress(kTestIpAddress1, kTestPort1)),
        net::NetworkAnonymizationKey());

    FakeDatagramServerSocket* datagram_socket =
        GetSocketFromHost(socket_impl.get());
    net::IPEndPoint bound_address;
    datagram_socket->GetLocalAddress(&bound_address);
    EXPECT_EQ(port, bound_address.port());

    base::RunLoop().RunUntilIdle();

    socket_client = socket_impl->ReleaseClientForTesting();
    socket_receiver = socket_impl->ReleaseReceiverForTesting();
  }

  std::unique_ptr<P2PSocketUdp> socket_impl(new P2PSocketUdp(
      &socket_delegate_, std::move(socket_client), std::move(socket_receiver),
      &throttler, TRAFFIC_ANNOTATION_FOR_TESTS,
      /*net_log=*/nullptr, std::move(fake_socket_factory), std::nullopt));
  net::IPEndPoint local_address = ParseAddress(kTestLocalIpAddress, 0);

  auto* socket_impl_ptr = socket_impl.get();
  socket_delegate_.ExpectDestruction(std::move(socket_impl));
  socket_impl_ptr->Init(
      local_address, min_port, max_port,
      P2PHostAndIPEndPoint(std::string(),
                           ParseAddress(kTestIpAddress1, kTestPort1)),
      net::NetworkAnonymizationKey());

  base::RunLoop().RunUntilIdle();

  EXPECT_TRUE(fake_client2.connection_error());
}

// Verify that we can open a UDP socket listening in a given port included in
// a given valid range.
TEST_F(P2PSocketUdpTest, PortRangeExplictValidPort) {
  const uint16_t min_port = 10000;
  const uint16_t max_port = 10001;
  const uint16_t valid_port = min_port;
  base::circular_deque<FakeDatagramServerSocket::UDPPacket> sent_packets;
  std::vector<uint16_t> used_ports;
  P2PSocketUdp::DatagramServerSocketFactory fake_socket_factory =
      base::BindRepeating(&CreateFakeDatagramServerSocket, &sent_packets,
                          &used_ports, &fake_clock_);
  P2PMessageThrottler throttler;

  mojo::PendingRemote<mojom::P2PSocketClient> socket_client;
  mojo::PendingRemote<mojom::P2PSocket> socket;
  auto socket_receiver = socket.InitWithNewPipeAndPassReceiver();

  FakeSocketClient fake_client2(std::move(socket),
                                socket_client.InitWithNewPipeAndPassReceiver());

  EXPECT_CALL(fake_client2, SocketCreated(_, _)).Times(1);

  std::unique_ptr<P2PSocketUdp> socket_host(new P2PSocketUdp(
      &socket_delegate_, std::move(socket_client), std::move(socket_receiver),
      &throttler, TRAFFIC_ANNOTATION_FOR_TESTS,
      /*net_log=*/nullptr, std::move(fake_socket_factory), std::nullopt));
  net::IPEndPoint local_address = ParseAddress(kTestLocalIpAddress, valid_port);
  socket_host->Init(
      local_address, min_port, max_port,
      P2PHostAndIPEndPoint(std::string(),
                           ParseAddress(kTestIpAddress1, kTestPort1)),
      net::NetworkAnonymizationKey());

  FakeDatagramServerSocket* fake_socket = GetSocketFromHost(socket_host.get());
  net::IPEndPoint bound_address;
  fake_socket->GetLocalAddress(&bound_address);
  EXPECT_EQ(local_address.port(), bound_address.port());

  base::RunLoop().RunUntilIdle();
}

// Verify that we cannot open a UDP socket listening in a given port not
// included in a given valid range.
TEST_F(P2PSocketUdpTest, PortRangeExplictInvalidPort) {
  const uint16_t min_port = 10000;
  const uint16_t max_port = 10001;
  const uint16_t invalid_port = max_port + 1;
  base::circular_deque<FakeDatagramServerSocket::UDPPacket> sent_packets;
  std::vector<uint16_t> used_ports;
  P2PSocketUdp::DatagramServerSocketFactory fake_socket_factory =
      base::BindRepeating(&CreateFakeDatagramServerSocket, &sent_packets,
                          &used_ports, &fake_clock_);
  P2PMessageThrottler throttler;

  mojo::PendingRemote<mojom::P2PSocketClient> socket_client;
  mojo::PendingRemote<mojom::P2PSocket> socket;
  auto socket_receiver = socket.InitWithNewPipeAndPassReceiver();

  FakeSocketClient fake_client2(std::move(socket),
                                socket_client.InitWithNewPipeAndPassReceiver());

  auto socket_impl = std::make_unique<P2PSocketUdp>(
      &socket_delegate_, std::move(socket_client), std::move(socket_receiver),
      &throttler, TRAFFIC_ANNOTATION_FOR_TESTS, /*net_log=*/nullptr,
      std::move(fake_socket_factory), std::nullopt);
  net::IPEndPoint local_address =
      ParseAddress(kTestLocalIpAddress, invalid_port);

  auto* socket_impl_ptr = socket_impl.get();
  socket_delegate_.ExpectDestruction(std::move(socket_impl));
  socket_impl_ptr->Init(
      local_address, min_port, max_port,
      P2PHostAndIPEndPoint(std::string(),
                           ParseAddress(kTestIpAddress1, kTestPort1)),
      net::NetworkAnonymizationKey());

  base::RunLoop().RunUntilIdle();

  EXPECT_TRUE(fake_client2.connection_error());
}

// Verify that we can receive packets from the sockets, and that the
// discontinuous packets are not batched.
TEST_F(P2PSocketUdpTest, ReceiveDiscontinuousPackets) {
  // Receive STUN request from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  base::RunLoop().RunUntilIdle();

  // Now we should be able to receive any data from |dest1_|.
  constexpr uint64_t kPacketIntervalNs =
      P2PSocketUdp::kUdpMaxBatchingRecvBuffering.InNanoseconds() / 2;

  std::vector<uint8_t> packet1;
  std::vector<uint8_t> packet2;
  std::vector<uint8_t> packet3;

  CreateRandomPacket(&packet1);
  CreateRandomPacket(&packet2);
  CreateRandomPacket(&packet3);

  InSequence s;
  // The socket returns `ERR_IO_PENDING` in between packets. It
  // indicates no more packets in the socket at that moment. The
  // packet1/packet2/packet3 are regarded as discontinuous.
  // Expect that the discontinuous packets are not batched.
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packet1), _));
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packet2), _));
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packet3), _));

  // Start to receive packets.
  socket_->ReceivePacket(dest1_, packet1);

  fake_clock_.SetTimeNanos(kPacketIntervalNs);
  socket_->ReceivePacket(dest1_, packet2);

  fake_clock_.SetTimeNanos(2 * kPacketIntervalNs);
  socket_->ReceivePacket(dest1_, packet3);

  base::RunLoop().RunUntilIdle();
}

// Verify that we can receive burst packets from the sockets, and that all the
// packets are batched together.
TEST_F(P2PSocketUdpTest, ReceiveBurstPacketsBasic) {
  // Receive STUN request from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  base::RunLoop().RunUntilIdle();

  // Now we should be able to receive any data from |dest1_|.
  constexpr size_t kNumPackets = P2PSocketUdp::kUdpMaxBatchingRecvPackets;

  std::vector<std::vector<uint8_t>> packets(kNumPackets);
  for (size_t i = 0; i < kNumPackets; i++) {
    CreateRandomPacket(&packets[i]);
    socket_->AddRecvPacket(dest1_, packets[i]);
  }

  InSequence s;
  // Expect to receive all the packets in one batching.
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  for (size_t i = 0; i < kNumPackets; i++) {
    EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packets[i]), _));
  }
  // Start to receive burst packets.
  socket_->FireRecvCallback();

  base::RunLoop().RunUntilIdle();
}

// Verify that we can receive burst packets, and that the batching size does not
// exceed limit.
TEST_F(P2PSocketUdpTest, ReceiveBurstPacketsExceedingMaxBatchingSize) {
  // Receive STUN request from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  base::RunLoop().RunUntilIdle();

  // Now we should be able to receive any data from |dest1_|.
  constexpr size_t kNumPacketsExceedingMaxBatching = 3;
  DCHECK_LE(kNumPacketsExceedingMaxBatching,
            P2PSocketUdp::kUdpMaxBatchingRecvPackets);
  constexpr size_t kNumPacketsAll = P2PSocketUdp::kUdpMaxBatchingRecvPackets +
                                    kNumPacketsExceedingMaxBatching;

  std::vector<std::vector<uint8_t>> packets(kNumPacketsAll);
  for (size_t i = 0; i < kNumPacketsAll; i++) {
    CreateRandomPacket(&packets[i]);
    socket_->AddRecvPacket(dest1_, packets[i]);
  }

  InSequence s;
  size_t i = 0;
  // Expect to receive maximum allowed number of packets in the first batching.
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  for (; i < P2PSocketUdp::kUdpMaxBatchingRecvPackets; i++) {
    EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packets[i]), _));
  }
  // Expect to receive the remainder packets in the second batching.
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  for (; i < kNumPacketsAll; i++) {
    EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packets[i]), _));
  }
  // Start to receive burst packets.
  socket_->FireRecvCallback();

  base::RunLoop().RunUntilIdle();
}

// Verify that we can receive burst packets, and that the batching cancels if
// buffering time exceeds limit.
TEST_F(P2PSocketUdpTest, ReceiveBurstPacketsExceedingMaxBatchingBuffering) {
  // Receive STUN request from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  base::RunLoop().RunUntilIdle();

  // Now we should be able to receive any data from |dest1_|.
  constexpr size_t kNumPacketsWithProcessLatency = 16;
  constexpr size_t kNumPacketsExceedingMaximumBuffering = 8;
  constexpr size_t kNumPacketsAll =
      kNumPacketsWithProcessLatency + kNumPacketsExceedingMaximumBuffering;
  DCHECK_LE(kNumPacketsAll, P2PSocketUdp::kUdpMaxBatchingRecvPackets);

  std::vector<std::vector<uint8_t>> packets(kNumPacketsAll);
  for (size_t i = 0; i < kNumPacketsAll; i++) {
    CreateRandomPacket(&packets[i]);
  }

  constexpr uint64_t kMaximumBatchingBufferingNs =
      P2PSocketUdp::kUdpMaxBatchingRecvBuffering.InNanoseconds();
  // Latency of `P2PSocketUdp` to retrieve one packet from socket.
  constexpr uint64_t kPacketProcessLatencyNs =
      kMaximumBatchingBufferingNs / kNumPacketsWithProcessLatency;

  // Add packets with process latency. The total latency does not exceed limit.
  for (size_t i = 0; i < kNumPacketsWithProcessLatency; i++) {
    socket_->AddRecvPacket(dest1_, packets[i], kPacketProcessLatencyNs * i);
  }
  // Add the packet with maximum buffering time plus 1 microsecond, which
  // immediately cancels batching more packets.
  socket_->AddRecvPacket(
      dest1_, packets[kNumPacketsWithProcessLatency],
      kMaximumBatchingBufferingNs + webrtc::kNumNanosecsPerMicrosec);
  // Add the remainder packets.
  for (size_t i = kNumPacketsWithProcessLatency + 1; i < kNumPacketsAll; i++) {
    socket_->AddRecvPacket(dest1_, packets[i]);
  }

  InSequence s;
  // Expect to receive the first batching packets.
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  for (size_t i = 0; i < kNumPacketsWithProcessLatency + 1; i++) {
    EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packets[i]), _));
  }
  // Expect to receive the remainder packets in the second batching.
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  for (size_t i = kNumPacketsWithProcessLatency + 1; i < kNumPacketsAll; i++) {
    EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packets[i]), _));
  }
  // Start to receive burst packets.
  socket_->FireRecvCallback();

  base::RunLoop().RunUntilIdle();
}

class P2PSocketUdpWithInterceptorTest : public P2PSocketUdpTest {
 public:
  P2PSocketUdpWithInterceptorTest()
      : P2PSocketUdpTest(base::UnguessableToken::Create()),
        throttling_client_id_(base::UnguessableToken::Create()),
        throttling_token_(
            ScopedThrottlingToken::MaybeCreate(net_log_with_source_.source().id,
                                               devtools_token_)) {}

  void SetUp() override {
    SetNetworkState({});
    P2PSocketUdpTest::SetUp();
  }

  void TearDown() override {
    RemoveThrottling();
    P2PSocketUdpTest::TearDown();
  }

  struct NetworkState {
    bool offline = false;
    base::TimeDelta latency;
    double packet_loss = 0.0;
    int packet_queue_length = 0;
  };

  void SetNetworkState(NetworkState state) {
    ThrottlingController::SetConditions(
        *devtools_token_, throttling_client_id_,
        {{{},
          NetworkConditions{state.offline, state.latency.InMillisecondsF(), 0.0,
                            0.0, state.packet_loss, state.packet_queue_length,
                            false, std::nullopt}}});
  }

  void RemoveThrottling() {
    ThrottlingController::SetConditions(*devtools_token_, throttling_client_id_,
                                        {});
  }

  void AdvanceClock(base::TimeDelta delta) {
    base::TimeDelta now = base::Nanoseconds(fake_clock_.TimeNanos());
    fake_clock_.SetTimeNanos((now + delta).InNanoseconds());
    task_environment_.FastForwardBy(delta);
  }

 protected:
  base::UnguessableToken throttling_client_id_;
  std::unique_ptr<network::ScopedThrottlingToken> throttling_token_;
};

TEST_F(P2PSocketUdpWithInterceptorTest, SendPacket) {
  base::HistogramTester histograms;

  // Receive packet from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  // Now we should be able to send any data to |dest1_|.
  EXPECT_CALL(*fake_client_.get(), SendComplete(_));

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));

  AdvanceClock(base::Milliseconds(100));

  ASSERT_EQ(1U, sent_packets_.size());
  ASSERT_EQ(dest1_, std::get<0>(sent_packets_[0]));
  histograms.ExpectUniqueSample("WebRTC.P2P.UDP.SendResult", net::OK, 1);
}

TEST_F(P2PSocketUdpWithInterceptorTest, SendPacketOffline) {
  // Receive packet from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  // Now we should be able to send any data to |dest1_|.
  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(2);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  SetNetworkState({.offline = true});
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));
  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(0U, sent_packets_.size());

  SetNetworkState({});
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 1));
  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(1U, sent_packets_.size());
}

TEST_F(P2PSocketUdpWithInterceptorTest, SendPacketDelayed) {
  // Receive packet from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  // Now we should be able to send any data to |dest1_|.
  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(2);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  SetNetworkState({.latency = base::Milliseconds(1000)});
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));

  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(0U, sent_packets_.size());

  AdvanceClock(base::Milliseconds(2000));
  EXPECT_EQ(1U, sent_packets_.size());

  SetNetworkState({});
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 1));

  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(2U, sent_packets_.size());
}

TEST_F(P2PSocketUdpWithInterceptorTest, SendPacketAndRemoveThrottling) {
  // Receive packet from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  // Now we should be able to send any data to |dest1_|.
  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(2);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));

  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(1U, sent_packets_.size());

  RemoveThrottling();

  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));

  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(2U, sent_packets_.size());
}

TEST_F(P2PSocketUdpWithInterceptorTest, SendPacketDropsLongQueue) {
  constexpr size_t kMaxQueueLength = 100;
  SetNetworkState({.packet_queue_length = kMaxQueueLength});

  // Receive packet from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  // Now we should be able to send any data to |dest1_|.
  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(500);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  for (int i = 0; i < 500; ++i) {
    socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));
  }

  AdvanceClock(base::Milliseconds(1000));
  EXPECT_EQ(kMaxQueueLength, sent_packets_.size());
}

TEST_F(P2PSocketUdpWithInterceptorTest, SendPacketWithPacketDrop) {
  // Receive packet from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(2);

  socket_->ReceivePacket(dest1_, request_packet);
  AdvanceClock(base::Milliseconds(100));

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  SetNetworkState({.packet_loss = 100.0});
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));
  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(0U, sent_packets_.size());

  SetNetworkState({});
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 1));
  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(1U, sent_packets_.size());
}

TEST_F(P2PSocketUdpWithInterceptorTest, ReceivePackets) {
  // Receive STUN request from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  constexpr size_t kNumPackets = P2PSocketUdp::kUdpMaxBatchingRecvPackets;
  std::vector<std::vector<uint8_t>> packets(kNumPackets);
  for (size_t i = 0; i < kNumPackets; i++) {
    CreateRandomPacket(&packets[i]);
  }

  InSequence s;
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  for (size_t i = 0; i < kNumPackets; i++) {
    EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
    EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packets[i]), _));
  }

  socket_->ReceivePacket(dest1_, request_packet);

  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(1U, received_packets_.size());

  // Now we should be able to receive any data from |dest1_|.
  for (size_t i = 0; i < kNumPackets; i++) {
    socket_->ReceivePacket(dest1_, packets[i]);
  }

  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(kNumPackets + 1U, received_packets_.size());
}

// Verify that we can receive Explicit Congestion Notification (ECN) bits
// from the socket after enabling the socket option, while assuming that
// the sender is sending the ECN bits.
TEST_F(P2PSocketUdpWithInterceptorTest, ReceivePacketsWithEcn) {
  // Receive STUN request from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  constexpr size_t kNumPackets = P2PSocketUdp::kUdpMaxBatchingRecvPackets;
  std::vector<std::vector<uint8_t>> packets(kNumPackets);
  for (size_t i = 0; i < kNumPackets; i++) {
    CreateRandomPacket(&packets[i]);
  }

  InSequence s;
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  for (size_t i = 0; i < kNumPackets; i++) {
    EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
    EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packets[i]), _));
  }

  int desired_recv_ecn = 1;
  socket_->ReceivePacket(dest1_, request_packet);

  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(1U, received_packets_.size());
  // Before setting the ECN receiving option on the socket,
  // it will return the default ECN bits.
  EXPECT_EQ(net::ECN_DEFAULT, socket_->GetLastTos().ecn);

  // Setting the ECN bits receiving option for the socket.
  socket_impl_->SetOption(P2P_SOCKET_OPT_RECV_ECN, desired_recv_ecn);
  // Now we should be able to receive any data from |dest1_| with the ECN bits.
  for (size_t i = 0; i < kNumPackets; i++) {
    socket_->ReceivePacket(dest1_, packets[i]);
    EXPECT_EQ(net::ECN_ECT1, socket_->GetLastTos().ecn);
  }

  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(kNumPackets + 1U, received_packets_.size());
}

TEST_F(P2PSocketUdpWithInterceptorTest, ReceivePacketDelayed) {
  // Receive STUN request from |dest1_|.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  InSequence s;
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  AdvanceClock(base::Milliseconds(100));

  SetNetworkState({.latency = base::Milliseconds(1000)});

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(0);
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);
  socket_->ReceivePacket(dest1_, packet);

  AdvanceClock(base::Milliseconds(100));
  EXPECT_EQ(1U, received_packets_.size());

  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(packet), _));
  AdvanceClock(base::Milliseconds(2000));
  EXPECT_EQ(2U, received_packets_.size());
}

TEST_F(P2PSocketUdpWithInterceptorTest, ReentrantDestructionSend) {
  // Enable throttling.
  SetNetworkState({.latency = base::Milliseconds(100)});

  // Send a packet, which starts the retry timer.
  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));

  // Mark the socket for destruction.
  socket_delegate_.ExpectDestruction(std::move(socket_impl_));
  socket_ = nullptr;

  // Advance clock past the retry delay, which will call OnSendNetworkTimer(),
  // DoSend() and DestroySocket() synchronously.
  AdvanceClock(base::Milliseconds(100));
}

// Verify that when `SendTo()` returns `ERR_NO_BUFFER_SPACE`, the packet is
// retried after a timer fires.
TEST_F(P2PSocketUdpTest, RetrySendAfterNoBufferSpace) {
  base::HistogramTester histograms;

  // Simulate an `ERR_NO_BUFFER_SPACE` failure.
  socket_->send_result_queue().push_back(net::ERR_NO_BUFFER_SPACE);

  // Receive packet from `dest1_` to allow sending data.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(1);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));

  // The packet must not have been sent yet due to `ERR_NO_BUFFER_SPACE`.
  ASSERT_EQ(0U, sent_packets_.size());

  // Advance time past the retry delay.
  task_environment_.FastForwardBy(base::Milliseconds(1));

  // After the retry timer fires, the packet must have been sent.
  ASSERT_EQ(1U, sent_packets_.size());
  ASSERT_EQ(dest1_, std::get<0>(sent_packets_[0]));

  histograms.ExpectUniqueSample("WebRTC.P2P.UDP.SendResult", net::OK, 1);
}

// Verify that queued packets are sent after a retry completes.
TEST_F(P2PSocketUdpTest, RetrySendThenSendQueuedPackets) {
  base::HistogramTester histograms;

  // Simulate an `ERR_NO_BUFFER_SPACE` failure.
  socket_->send_result_queue().push_back(net::ERR_NO_BUFFER_SPACE);

  // Receive packet from `dest1_` to allow sending data.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(0);

  // The async send callback, `P2PSocketUdp::OnSend()`, sends the second packet
  // and then runs both send completion callbacks.
  EXPECT_CALL(*fake_client_.get(), SendBatchComplete(_)).Times(1);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet1;
  CreateRandomPacket(&packet1);
  std::vector<uint8_t> packet2;
  CreateRandomPacket(&packet2);

  // First packet triggers a retry. Second is queued because the first send is
  // pending.
  socket_impl_->Send(packet1, P2PPacketInfo(dest1_, options, 0));
  socket_impl_->Send(packet2, P2PPacketInfo(dest1_, options, 1));

  ASSERT_EQ(0U, sent_packets_.size());

  // Advance past the retry delay.
  task_environment_.FastForwardBy(base::Milliseconds(1));

  // Both packets should have been sent.
  ASSERT_EQ(2U, sent_packets_.size());
  ASSERT_EQ(std::get<1>(sent_packets_[0]), packet1);
  ASSERT_EQ(std::get<1>(sent_packets_[1]), packet2);

  histograms.ExpectUniqueSample("WebRTC.P2P.UDP.SendResult", net::OK, 2);
}

// Verify exponential backoff doubles the delay for each retry.
TEST_F(P2PSocketUdpTest, RetrySendWithExponentialBackoff) {
  base::HistogramTester histograms;

  // Simulate repeated `ERR_NO_BUFFER_SPACE` failures.
  for (int i = 0; i < 3; ++i) {
    socket_->send_result_queue().push_back(net::ERR_NO_BUFFER_SPACE);
  }

  // Receive packet from `dest1_` to allow sending data.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(1);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));
  ASSERT_EQ(0U, sent_packets_.size());

  // The first retry must fail.
  task_environment_.FastForwardBy(base::Milliseconds(1));
  ASSERT_EQ(0U, sent_packets_.size());

  // The second retry must fail.
  task_environment_.FastForwardBy(base::Milliseconds(2));
  ASSERT_EQ(0U, sent_packets_.size());

  // The third retry must succeed.
  task_environment_.FastForwardBy(base::Milliseconds(4));
  ASSERT_EQ(1U, sent_packets_.size());

  histograms.ExpectUniqueSample("WebRTC.P2P.UDP.SendResult", net::OK, 1);
}

// Verify that the socket is destroyed and a connection error is reported when
// all send retries are exhausted due to repeated `ERR_NO_BUFFER_SPACE` errors.
TEST_F(P2PSocketUdpTest, RetrySendWithTimeout) {
  base::HistogramTester histograms;

  // Simulate repeated `ERR_NO_BUFFER_SPACE` failures.
  for (size_t i = 0; i < P2PSocketUdp::kMaxSendRetries + 1; ++i) {
    socket_->send_result_queue().push_back(net::ERR_NO_BUFFER_SPACE);
  }

  // Receive packet from `dest1_` to allow sending data.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  socket_ = nullptr;
  auto* socket_impl_ptr = socket_impl_.get();
  socket_delegate_.ExpectDestruction(std::move(socket_impl_));
  socket_impl_ptr->Send(packet, P2PPacketInfo(dest1_, options, 0));
  ASSERT_EQ(0U, sent_packets_.size());

  base::TimeDelta total_delay =
      base::Milliseconds(std::pow(2, P2PSocketUdp::kMaxSendRetries) - 1);
  task_environment_.FastForwardBy(total_delay);

  ASSERT_EQ(0U, sent_packets_.size());
  EXPECT_TRUE(fake_client_->connection_error());

  histograms.ExpectUniqueSample("WebRTC.P2P.UDP.SendResult",
                                net::ERR_NO_BUFFER_SPACE, 1);
}

// Verify that a non-retryable error during a retry attempt causes the
// socket to be destroyed immediately instead of continuing to retry.
TEST_F(P2PSocketUdpTest, RetrySendWithError) {
  base::HistogramTester histograms;

  // Simulate a retryable failure followed by a non-retryable failure.
  socket_->send_result_queue().push_back(net::ERR_NO_BUFFER_SPACE);
  socket_->send_result_queue().push_back(net::ERR_FAILED);

  // Receive packet from `dest1_` to allow sending data.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(0);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);

  socket_ = nullptr;
  auto* socket_impl_ptr = socket_impl_.get();
  socket_delegate_.ExpectDestruction(std::move(socket_impl_));
  socket_impl_ptr->Send(packet, P2PPacketInfo(dest1_, options, 0));

  // The packet must not have been sent yet due to `ERR_NO_BUFFER_SPACE`.
  ASSERT_EQ(0U, sent_packets_.size());

  // Advance time past the retry delay.
  task_environment_.FastForwardBy(base::Milliseconds(1));

  // The socket must have been destroyed due to the error.
  EXPECT_TRUE(fake_client_->connection_error());
  ASSERT_EQ(0U, sent_packets_.size());

  histograms.ExpectUniqueSample("WebRTC.P2P.UDP.SendResult", net::ERR_FAILED,
                                1);
}

// Verify that a retry followed by an asynchronous send completes successfully.
TEST_F(P2PSocketUdpTest, RetrySendThenAsyncSend) {
  base::HistogramTester histograms;

  // Simulate a retry error followed by an async send.
  socket_->send_result_queue().push_back(net::ERR_NO_BUFFER_SPACE);
  socket_->send_result_queue().push_back(net::ERR_IO_PENDING);

  // Receive packet from `dest1_` to allow sending data.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(1);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _));
  socket_->ReceivePacket(dest1_, request_packet);

  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(1);

  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));

  // The packet must not have been sent yet due to `ERR_NO_BUFFER_SPACE`.
  ASSERT_EQ(0U, sent_packets_.size());

  // Advance time past the retry delay.
  task_environment_.FastForwardBy(base::Milliseconds(1));

  // After the retry timer fires, the packet must have been sent.
  ASSERT_EQ(1U, sent_packets_.size());
  ASSERT_EQ(dest1_, std::get<0>(sent_packets_[0]));

  histograms.ExpectUniqueSample("WebRTC.P2P.UDP.SendResult", net::OK, 1);
}

// Verify that a pending send retry is cancelled when a read error destroys the
// socket before the retry timer fires.
TEST_F(P2PSocketUdpTest, RetrySendCancelledByReadError) {
  base::HistogramTester histograms;

  // Simulate an `ERR_NO_BUFFER_SPACE` failure.
  socket_->send_result_queue().push_back(net::ERR_NO_BUFFER_SPACE);

  // Receive packet from `dest1_` to allow sending data.
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);
  EXPECT_CALL(*fake_client_.get(), DataReceived(_)).Times(2);
  EXPECT_CALL(*this, SinglePacketReceptionHelper(_, SpanEq(request_packet), _))
      .Times(2);
  socket_->ReceivePacket(dest1_, request_packet);

  EXPECT_CALL(*fake_client_.get(), SendComplete(_)).Times(0);

  // Send a packet, which starts the retry timer.
  webrtc::AsyncSocketPacketOptions options;
  std::vector<uint8_t> packet;
  CreateRandomPacket(&packet);
  socket_impl_->Send(packet, P2PPacketInfo(dest1_, options, 0));

  // The packet must not have been sent yet due to `ERR_NO_BUFFER_SPACE`.
  ASSERT_EQ(0U, sent_packets_.size());

  // Simulate a read error, destroying the socket.
  socket_->SetReadErrorResult(net::ERR_FAILED);
  socket_delegate_.ExpectDestruction(std::move(socket_impl_));

  auto* socket_ptr = socket_.get();
  socket_ = nullptr;
  socket_ptr->ReceivePacket(dest1_, request_packet);

  // Advance time past the retry delay.
  task_environment_.FastForwardBy(base::Milliseconds(1));

  // After the retry timer fires, the packet must not have sent.
  ASSERT_EQ(0U, sent_packets_.size());

  histograms.ExpectTotalCount("WebRTC.P2P.UDP.SendResult", 0);
}

TEST_F(P2PSocketUdpTest, SendRejectsRestrictedPort) {
  base::test::ScopedFeatureList feature_list;
  feature_list.InitAndEnableFeature(kEnforceP2PSocketPortRestrictions);
  std::vector<uint8_t> request_packet;
  CreateStunRequest(&request_packet);

  net::IPEndPoint restricted_dest = ParseAddress(kTestIpAddress1, 25);

  socket_ = nullptr;
  P2PSocketUdp* socket_impl_ptr = socket_impl_.get();
  socket_delegate_.ExpectDestruction(std::move(socket_impl_));
  socket_impl_ptr->Send(request_packet, P2PPacketInfo(restricted_dest, {}, 0));

  EXPECT_TRUE(
      base::test::RunUntil([&]() { return fake_client_->connection_error(); }));
}

}  // namespace network