dcsctp: Add Heartbeat Handler

It's responsible for answering incoming Heartbeat Requests, and to
send requests itself when a connection is idle. When it receives
a response, it will measure the RTT and if it doesn't receive a response
in time, that will result in a TX error, which will eventually close
the connection.

Bug: webrtc:12614
Change-Id: I08371d9072ff0461f60e0a2f7696c0fd7ccb57c5
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/214129
Reviewed-by: Tommi <tommi@webrtc.org>
Commit-Queue: Victor Boivie <boivie@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#33828}

net/dcsctp/BUILD.gn

@@ -17,6 +17,7 @@ if (rtc_include_tests) {
       "packet:dcsctp_packet_unittests",
       "public:dcsctp_public_unittests",
       "rx:dcsctp_rx_unittests",
+      "socket:dcsctp_socket_unittests",
       "timer:dcsctp_timer_unittests",
       "tx:dcsctp_tx_unittests",
     ]

net/dcsctp/socket/BUILD.gn

@@ -18,6 +18,21 @@ rtc_source_set("context") {
   ]
 }
 
+rtc_library("heartbeat_handler") {
+  deps = [
+    ":context",
+    "../../../api:array_view",
+    "../../../rtc_base",
+    "../../../rtc_base:checks",
+    "../../../rtc_base:rtc_base_approved",
+    "../public:types",
+  ]
+  sources = [
+    "heartbeat_handler.cc",
+    "heartbeat_handler.h",
+  ]
+}
+
 if (rtc_include_tests) {
   rtc_source_set("mock_callbacks") {
     testonly = true
@@ -37,4 +52,18 @@ if (rtc_include_tests) {
       "../public:types",
     ]
   }
+
+  rtc_library("dcsctp_socket_unittests") {
+    testonly = true
+
+    deps = [
+      ":heartbeat_handler",
+      "../../../api:array_view",
+      "../../../rtc_base:checks",
+      "../../../rtc_base:gunit_helpers",
+      "../../../rtc_base:rtc_base_approved",
+      "../../../test:test_support",
+    ]
+    sources = [ "heartbeat_handler_test.cc" ]
+  }
 }

net/dcsctp/socket/heartbeat_handler.cc

@@ -0,0 +1,189 @@
+/*
+ *  Copyright (c) 2021 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+#include "net/dcsctp/socket/heartbeat_handler.h"
+
+#include <stddef.h>
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "absl/strings/string_view.h"
+#include "absl/types/optional.h"
+#include "api/array_view.h"
+#include "net/dcsctp/packet/bounded_byte_reader.h"
+#include "net/dcsctp/packet/bounded_byte_writer.h"
+#include "net/dcsctp/packet/chunk/heartbeat_ack_chunk.h"
+#include "net/dcsctp/packet/chunk/heartbeat_request_chunk.h"
+#include "net/dcsctp/packet/parameter/heartbeat_info_parameter.h"
+#include "net/dcsctp/packet/parameter/parameter.h"
+#include "net/dcsctp/packet/sctp_packet.h"
+#include "net/dcsctp/public/dcsctp_options.h"
+#include "net/dcsctp/public/dcsctp_socket.h"
+#include "net/dcsctp/socket/context.h"
+#include "net/dcsctp/timer/timer.h"
+#include "rtc_base/logging.h"
+
+namespace dcsctp {
+
+// This is stored (in serialized form) as HeartbeatInfoParameter sent in
+// HeartbeatRequestChunk and received back in HeartbeatAckChunk. It should be
+// well understood that this data may be modified by the peer, so it can't
+// be trusted.
+//
+// It currently only stores a timestamp, in millisecond precision, to allow for
+// RTT measurements. If that would be manipulated by the peer, it would just
+// result in incorrect RTT measurements, which isn't an issue.
+class HeartbeatInfo {
+ public:
+  static constexpr size_t kBufferSize = sizeof(uint64_t);
+  static_assert(kBufferSize == 8, "Unexpected buffer size");
+
+  explicit HeartbeatInfo(TimeMs created_at) : created_at_(created_at) {}
+
+  std::vector<uint8_t> Serialize() {
+    uint32_t high_bits = static_cast<uint32_t>(*created_at_ >> 32);
+    uint32_t low_bits = static_cast<uint32_t>(*created_at_);
+
+    std::vector<uint8_t> data(kBufferSize);
+    BoundedByteWriter<kBufferSize> writer(data);
+    writer.Store32<0>(high_bits);
+    writer.Store32<4>(low_bits);
+    return data;
+  }
+
+  static absl::optional<HeartbeatInfo> Deserialize(
+      rtc::ArrayView<const uint8_t> data) {
+    if (data.size() != kBufferSize) {
+      RTC_LOG(LS_WARNING) << "Invalid heartbeat info: " << data.size()
+                          << " bytes";
+      return absl::nullopt;
+    }
+
+    BoundedByteReader<kBufferSize> reader(data);
+    uint32_t high_bits = reader.Load32<0>();
+    uint32_t low_bits = reader.Load32<4>();
+
+    uint64_t created_at = static_cast<uint64_t>(high_bits) << 32 | low_bits;
+    return HeartbeatInfo(TimeMs(created_at));
+  }
+
+  TimeMs created_at() const { return created_at_; }
+
+ private:
+  const TimeMs created_at_;
+};
+
+HeartbeatHandler::HeartbeatHandler(absl::string_view log_prefix,
+                                   const DcSctpOptions& options,
+                                   Context* context,
+                                   TimerManager* timer_manager)
+    : log_prefix_(std::string(log_prefix) + "heartbeat: "),
+      ctx_(context),
+      timer_manager_(timer_manager),
+      interval_duration_(options.heartbeat_interval),
+      interval_duration_should_include_rtt_(
+          options.heartbeat_interval_include_rtt),
+      interval_timer_(timer_manager_->CreateTimer(
+          "heartbeat-interval",
+          [this]() { return OnIntervalTimerExpiry(); },
+          TimerOptions(interval_duration_, TimerBackoffAlgorithm::kFixed))),
+      timeout_timer_(timer_manager_->CreateTimer(
+          "heartbeat-timeout",
+          [this]() { return OnTimeoutTimerExpiry(); },
+          TimerOptions(options.rto_initial,
+                       TimerBackoffAlgorithm::kExponential,
+                       /*max_restarts=*/0))) {
+  // The interval timer must always be running as long as the association is up.
+  interval_timer_->Start();
+}
+
+void HeartbeatHandler::RestartTimer() {
+  if (interval_duration_should_include_rtt_) {
+    // The RTT should be used, but it's not easy accessible. The RTO will
+    // suffice.
+    interval_timer_->set_duration(interval_duration_ + ctx_->current_rto());
+  } else {
+    interval_timer_->set_duration(interval_duration_);
+  }
+
+  interval_timer_->Start();
+}
+
+void HeartbeatHandler::HandleHeartbeatRequest(HeartbeatRequestChunk chunk) {
+  // https://tools.ietf.org/html/rfc4960#section-8.3
+  // "The receiver of the HEARTBEAT should immediately respond with a
+  // HEARTBEAT ACK that contains the Heartbeat Information TLV, together with
+  // any other received TLVs, copied unchanged from the received HEARTBEAT
+  // chunk."
+  ctx_->Send(ctx_->PacketBuilder().Add(
+      HeartbeatAckChunk(std::move(chunk).extract_parameters())));
+}
+
+void HeartbeatHandler::HandleHeartbeatAck(HeartbeatAckChunk chunk) {
+  timeout_timer_->Stop();
+  absl::optional<HeartbeatInfoParameter> info_param = chunk.info();
+  if (!info_param.has_value()) {
+    ctx_->callbacks().OnError(
+        ErrorKind::kParseFailed,
+        "Failed to parse HEARTBEAT-ACK; No Heartbeat Info parameter");
+    return;
+  }
+  absl::optional<HeartbeatInfo> info =
+      HeartbeatInfo::Deserialize(info_param->info());
+  if (!info.has_value()) {
+    ctx_->callbacks().OnError(ErrorKind::kParseFailed,
+                              "Failed to parse HEARTBEAT-ACK; Failed to "
+                              "deserialized Heartbeat info parameter");
+    return;
+  }
+
+  DurationMs duration(*ctx_->callbacks().TimeMillis() - *info->created_at());
+
+  ctx_->ObserveRTT(duration);
+
+  // https://tools.ietf.org/html/rfc4960#section-8.1
+  // "The counter shall be reset each time ... a HEARTBEAT ACK is received from
+  // the peer endpoint."
+  ctx_->ClearTxErrorCounter();
+}
+
+absl::optional<DurationMs> HeartbeatHandler::OnIntervalTimerExpiry() {
+  if (ctx_->is_connection_established()) {
+    HeartbeatInfo info(ctx_->callbacks().TimeMillis());
+    timeout_timer_->set_duration(ctx_->current_rto());
+    timeout_timer_->Start();
+    RTC_DLOG(LS_INFO) << log_prefix_ << "Sending HEARTBEAT with timeout "
+                      << *timeout_timer_->duration();
+
+    Parameters parameters = Parameters::Builder()
+                                .Add(HeartbeatInfoParameter(info.Serialize()))
+                                .Build();
+
+    ctx_->Send(ctx_->PacketBuilder().Add(
+        HeartbeatRequestChunk(std::move(parameters))));
+  } else {
+    RTC_DLOG(LS_VERBOSE)
+        << log_prefix_
+        << "Will not send HEARTBEAT when connection not established";
+  }
+  return absl::nullopt;
+}
+
+absl::optional<DurationMs> HeartbeatHandler::OnTimeoutTimerExpiry() {
+  // Note that the timeout timer is not restarted. It will be started again when
+  // the interval timer expires.
+  RTC_DCHECK(!timeout_timer_->is_running());
+  ctx_->IncrementTxErrorCounter("HEARTBEAT timeout");
+  return absl::nullopt;
+}
+}  // namespace dcsctp

net/dcsctp/socket/heartbeat_handler.h

@@ -0,0 +1,69 @@
+/*
+ *  Copyright (c) 2021 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+#ifndef NET_DCSCTP_SOCKET_HEARTBEAT_HANDLER_H_
+#define NET_DCSCTP_SOCKET_HEARTBEAT_HANDLER_H_
+
+#include <stdint.h>
+
+#include <memory>
+#include <string>
+
+#include "absl/strings/string_view.h"
+#include "net/dcsctp/packet/chunk/heartbeat_ack_chunk.h"
+#include "net/dcsctp/packet/chunk/heartbeat_request_chunk.h"
+#include "net/dcsctp/packet/sctp_packet.h"
+#include "net/dcsctp/public/dcsctp_options.h"
+#include "net/dcsctp/socket/context.h"
+#include "net/dcsctp/timer/timer.h"
+
+namespace dcsctp {
+
+// HeartbeatHandler handles all logic around sending heartbeats and receiving
+// the responses, as well as receiving incoming heartbeat requests.
+//
+// Heartbeats are sent on idle connections to ensure that the connection is
+// still healthy and to measure the RTT. If a number of heartbeats time out,
+// the connection will eventually be closed.
+class HeartbeatHandler {
+ public:
+  HeartbeatHandler(absl::string_view log_prefix,
+                   const DcSctpOptions& options,
+                   Context* context,
+                   TimerManager* timer_manager);
+
+  // Called when the heartbeat interval timer should be restarted. This is
+  // generally done every time data is sent, which makes the timer expire when
+  // the connection is idle.
+  void RestartTimer();
+
+  // Called on received HeartbeatRequestChunk chunks.
+  void HandleHeartbeatRequest(HeartbeatRequestChunk chunk);
+
+  // Called on received HeartbeatRequestChunk chunks.
+  void HandleHeartbeatAck(HeartbeatAckChunk chunk);
+
+ private:
+  absl::optional<DurationMs> OnIntervalTimerExpiry();
+  absl::optional<DurationMs> OnTimeoutTimerExpiry();
+
+  const std::string log_prefix_;
+  Context* ctx_;
+  TimerManager* timer_manager_;
+  // The time for a connection to be idle before a heartbeat is sent.
+  const DurationMs interval_duration_;
+  // Adding RTT to the duration will add some jitter, which is good in
+  // production, but less good in unit tests, which is why it can be disabled.
+  const bool interval_duration_should_include_rtt_;
+  const std::unique_ptr<Timer> interval_timer_;
+  const std::unique_ptr<Timer> timeout_timer_;
+};
+}  // namespace dcsctp
+
+#endif  // NET_DCSCTP_SOCKET_HEARTBEAT_HANDLER_H_

net/dcsctp/socket/heartbeat_handler_test.cc

@@ -0,0 +1,123 @@
+/*
+ *  Copyright (c) 2021 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+#include "net/dcsctp/socket/heartbeat_handler.h"
+
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "net/dcsctp/packet/chunk/heartbeat_ack_chunk.h"
+#include "net/dcsctp/packet/chunk/heartbeat_request_chunk.h"
+#include "net/dcsctp/packet/parameter/heartbeat_info_parameter.h"
+#include "net/dcsctp/public/types.h"
+#include "net/dcsctp/socket/mock_context.h"
+#include "net/dcsctp/testing/testing_macros.h"
+#include "rtc_base/gunit.h"
+#include "test/gmock.h"
+
+namespace dcsctp {
+namespace {
+using ::testing::ElementsAre;
+using ::testing::IsEmpty;
+using ::testing::NiceMock;
+using ::testing::Return;
+using ::testing::SizeIs;
+
+DcSctpOptions MakeOptions() {
+  DcSctpOptions options;
+  options.heartbeat_interval_include_rtt = false;
+  options.heartbeat_interval = DurationMs(30'000);
+  return options;
+}
+
+class HeartbeatHandlerTest : public testing::Test {
+ protected:
+  HeartbeatHandlerTest()
+      : options_(MakeOptions()),
+        context_(&callbacks_),
+        timer_manager_([this]() { return callbacks_.CreateTimeout(); }),
+        handler_("log: ", options_, &context_, &timer_manager_) {}
+
+  const DcSctpOptions options_;
+  NiceMock<MockDcSctpSocketCallbacks> callbacks_;
+  NiceMock<MockContext> context_;
+  TimerManager timer_manager_;
+  HeartbeatHandler handler_;
+};
+
+TEST_F(HeartbeatHandlerTest, RepliesToHeartbeatRequests) {
+  uint8_t info_data[] = {1, 2, 3, 4, 5};
+  HeartbeatRequestChunk request(
+      Parameters::Builder().Add(HeartbeatInfoParameter(info_data)).Build());
+
+  handler_.HandleHeartbeatRequest(std::move(request));
+
+  std::vector<uint8_t> payload = callbacks_.ConsumeSentPacket();
+  ASSERT_HAS_VALUE_AND_ASSIGN(SctpPacket packet, SctpPacket::Parse(payload));
+  ASSERT_THAT(packet.descriptors(), SizeIs(1));
+
+  ASSERT_HAS_VALUE_AND_ASSIGN(
+      HeartbeatAckChunk response,
+      HeartbeatAckChunk::Parse(packet.descriptors()[0].data));
+
+  ASSERT_HAS_VALUE_AND_ASSIGN(
+      HeartbeatInfoParameter param,
+      response.parameters().get<HeartbeatInfoParameter>());
+
+  EXPECT_THAT(param.info(), ElementsAre(1, 2, 3, 4, 5));
+}
+
+TEST_F(HeartbeatHandlerTest, SendsHeartbeatRequestsOnIdleChannel) {
+  callbacks_.AdvanceTime(options_.heartbeat_interval);
+  for (TimeoutID id : callbacks_.RunTimers()) {
+    timer_manager_.HandleTimeout(id);
+  }
+
+  // Grab the request, and make a response.
+  std::vector<uint8_t> payload = callbacks_.ConsumeSentPacket();
+  ASSERT_HAS_VALUE_AND_ASSIGN(SctpPacket packet, SctpPacket::Parse(payload));
+  ASSERT_THAT(packet.descriptors(), SizeIs(1));
+
+  ASSERT_HAS_VALUE_AND_ASSIGN(
+      HeartbeatRequestChunk req,
+      HeartbeatRequestChunk::Parse(packet.descriptors()[0].data));
+
+  HeartbeatAckChunk ack(std::move(req).extract_parameters());
+
+  // Respond a while later. This RTT will be measured by the handler
+  constexpr DurationMs rtt(313);
+
+  EXPECT_CALL(context_, ObserveRTT(rtt)).Times(1);
+
+  callbacks_.AdvanceTime(rtt);
+  handler_.HandleHeartbeatAck(std::move(ack));
+}
+
+TEST_F(HeartbeatHandlerTest, IncreasesErrorIfNotAckedInTime) {
+  callbacks_.AdvanceTime(options_.heartbeat_interval);
+
+  DurationMs rto(105);
+  EXPECT_CALL(context_, current_rto).WillOnce(Return(rto));
+  for (TimeoutID id : callbacks_.RunTimers()) {
+    timer_manager_.HandleTimeout(id);
+  }
+
+  // Validate that a request was sent.
+  EXPECT_THAT(callbacks_.ConsumeSentPacket(), Not(IsEmpty()));
+
+  EXPECT_CALL(context_, IncrementTxErrorCounter).Times(1);
+  callbacks_.AdvanceTime(rto);
+  for (TimeoutID id : callbacks_.RunTimers()) {
+    timer_manager_.HandleTimeout(id);
+  }
+}
+
+}  // namespace
+}  // namespace dcsctp