fetch.h

Go to the documentation of this file.

// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the Apache 2.0 License.
#pragma once
 
#include "ccf/ds/nonstd.h"
#include "ccf/rest_verb.h"
#include "ds/internal_logger.h"
#include "http/curl.h"
#include "http/http_builder.h"
 
#include <charconv>
#include <curl/curl.h>
#include <llhttp/llhttp.h>
#include <memory>
#include <optional>
#include <span>
#include <stdexcept>
#include <string>
#include <vector>
 
#define EXPECT_HTTP_RESPONSE_STATUS( \
  request, status_code, expected, response_body) \
  do \
  { \
    if (status_code != expected) \
    { \
      std::string error_message; \
      /* Only include response body for 4xx and 5xx status codes */ \
      if (status_code >= 400 && status_code < 600) \
      { \
        std::string response_content; \
        if (response_body != nullptr) \
        { \
          response_content.assign( \
            response_body->buffer.begin(), response_body->buffer.end()); \
        } \
        else \
        { \
          response_content = "(no response body)"; \
        } \
        error_message = fmt::format( \
          "Expected {} response from {} {}, instead received {} ({})", \
          ccf::http_status_str(expected), \
          request->get_method().c_str(), \
          request->get_url(), \
          status_code, \
          response_content); \
      } \
      else \
      { \
        error_message = fmt::format( \
          "Expected {} response from {} {}, instead received {}", \
          ccf::http_status_str(expected), \
          request->get_method().c_str(), \
          request->get_url(), \
          status_code); \
      } \
      throw std::runtime_error(error_message); \
    } \
  } while (0)
 
namespace snapshots
{
  struct SnapshotResponse
  {
    std::string snapshot_name;
    std::vector<uint8_t> snapshot_data;
  };
 
  struct ContentRangeHeader
  {
    size_t range_start;
    size_t inclusive_range_end;
    size_t total_size;
  };
 
  static ContentRangeHeader parse_content_range_header(
    const ccf::curl::CurlRequest& request)
  {
    const auto& headers = request.get_response_headers();
 
    auto it = headers.find(ccf::http::headers::CONTENT_RANGE);
    if (it == headers.end())
    {
      throw std::runtime_error(
        "Response is missing expected content-range header");
    }
 
    auto [unit, remaining] = ccf::nonstd::split_1(it->second, " ");
    if (unit != "bytes")
    {
      throw std::runtime_error(
        "Unexpected content-range unit. Only 'bytes' is supported");
    }
 
    auto [range, total_size] = ccf::nonstd::split_1(remaining, "/");
    auto [range_start, range_end] = ccf::nonstd::split_1(range, "-");
 
    if (range_start.empty() || range_end.empty() || total_size.empty())
    {
      throw std::runtime_error(fmt::format(
        "Unsupported content-range header format. Expected 'bytes "
        "<begin>-<end>/<total>', received: {}",
        it->second));
    }
 
    ContentRangeHeader parsed_values{};
 
    {
      const auto [p, ec] = std::from_chars(
        range_start.begin(), range_start.end(), parsed_values.range_start);
      if (ec != std::errc())
      {
        throw std::runtime_error(fmt::format(
          "Could not parse range start ({}) from content-range header: {}",
          range_start,
          it->second));
      }
    }
 
    {
      const auto [p, ec] = std::from_chars(
        range_end.begin(), range_end.end(), parsed_values.inclusive_range_end);
      if (ec != std::errc())
      {
        throw std::runtime_error(fmt::format(
          "Could not parse range end ({}) from content-range header: {}",
          range_end,
          it->second));
      }
    }
 
    {
      const auto [p, ec] = std::from_chars(
        total_size.begin(), total_size.end(), parsed_values.total_size);
      if (ec != std::errc())
      {
        throw std::runtime_error(fmt::format(
          "Could not parse total size ({}) from content-range header: {}",
          total_size,
          it->second));
      }
    }
 
    {
      // Use content-length to determine whether the sender used an exclusive or
      // inclusive range end
      auto length_it = headers.find(ccf::http::headers::CONTENT_LENGTH);
      if (length_it == headers.end())
      {
        throw std::runtime_error(
          "Response is missing expected content-length header");
      }
 
      size_t content_length = 0;
 
      {
        const auto& length_s = length_it->second;
 
        const auto [p, ec] = std::from_chars(
          length_s.data(), length_s.data() + length_s.size(), content_length);
 
        if (ec != std::errc())
        {
          throw std::runtime_error(fmt::format(
            "Could not parse length from content-length header: {}",
            length_it->second));
        }
      }
 
      const auto range_length =
        parsed_values.inclusive_range_end - parsed_values.range_start;
      if (range_length == content_length)
      {
        LOG_INFO_FMT(
          "Server sent an exclusive-end content-range header. "
          "content-length={}, content-range={}. Adjusting this to local "
          "inclusive-end representation. This should be a temporary "
          "mismatch, between 6.x and 7.x nodes in a mixed network",
          length_it->second,
          it->second);
        parsed_values.inclusive_range_end -= 1;
      }
      else if (range_length + 1 == content_length)
      {
        LOG_DEBUG_FMT(
          "Server sent an inclusive-end content-range header. "
          "content-length={}, content-range={}. This is expected for 7.x to "
          "7.x nodes",
          length_it->second,
          it->second);
      }
      else
      {
        throw std::runtime_error(fmt::format(
          "content-range ({}, {} bytes) and content-length ({}) headers do not "
          "agree",
          it->second,
          range_length + 1,
          length_it->second));
      }
    }
 
    return parsed_values;
  }
 
  static std::optional<SnapshotResponse> try_fetch_from_peer(
    const std::string& peer_address,
    const std::vector<uint8_t>& peer_ca,
    size_t max_size,
    std::optional<size_t> since_seqno = std::nullopt)
  {
    try
    {
      ccf::curl::UniqueCURL curl_easy;
      curl_easy.set_blob_opt(
        CURLOPT_CAINFO_BLOB, peer_ca.data(), peer_ca.size());
 
      auto response_body = std::make_unique<ccf::curl::ResponseBody>(max_size);
 
      // Get snapshot. This may be redirected multiple times, and we follow
      // these redirects ourself so we can extract the final URL. Once the
      // redirects terminate, the final response is likely to be extremely
      // large so is fetched over multiple requests for a sub-range, returning
      // PARTIAL_CONTENT each time.
      std::string snapshot_url;
      if (since_seqno.has_value())
      {
        snapshot_url = fmt::format(
          "https://{}/node/snapshot?since={}", peer_address, *since_seqno);
      }
      else
      {
        snapshot_url = fmt::format("https://{}/node/snapshot", peer_address);
      }
 
      // Fetch 4MB chunks at a time
      constexpr size_t range_size = 4L * 1024 * 1024;
      size_t range_start = 0;
      size_t range_end = range_size;
      size_t inclusive_range_end = range_end - 1;
      bool fetched_all = false;
 
      auto process_partial_response =
        [&](const ccf::curl::CurlRequest& request) {
          auto content_range = parse_content_range_header(request);
 
          if (content_range.range_start != range_start)
          {
            throw std::runtime_error(fmt::format(
              "Unexpected range response. Requested bytes {}-{}, received "
              "range starting at {}",
              range_start,
              range_end,
              content_range.range_start));
          }
 
          // The server may give us _less_ than we requested (since they know
          // where the file ends), but should never give us more
          if (content_range.inclusive_range_end > inclusive_range_end)
          {
            throw std::runtime_error(fmt::format(
              "Unexpected range response. Requested bytes {}-{}, received "
              "range ending at {}",
              range_start,
              inclusive_range_end,
              content_range.inclusive_range_end));
          }
 
          const auto content_range_exclusive_range_end =
            content_range.inclusive_range_end + 1;
 
          const auto range_size =
            content_range_exclusive_range_end - content_range.range_start;
          LOG_TRACE_FMT(
            "Received {}-byte chunk from {}. Now have {}/{}",
            range_size,
            request.get_url(),
            content_range_exclusive_range_end,
            content_range.total_size);
 
          if (content_range_exclusive_range_end == content_range.total_size)
          {
            fetched_all = true;
          }
          else
          {
            // Advance range for next request
            range_start = range_end;
            range_end = range_start + range_size;
            inclusive_range_end = range_end - 1;
          }
        };
 
      const auto max_redirects = 20;
      for (auto redirect_count = 1; redirect_count <= max_redirects;
           ++redirect_count)
      {
        LOG_DEBUG_FMT(
          "Making snapshot discovery request {}/{} to {}",
          redirect_count,
          max_redirects,
          snapshot_url);
 
        ccf::curl::UniqueSlist headers;
        headers.append(
          ccf::http::headers::RANGE,
          fmt::format("bytes={}-{}", range_start, inclusive_range_end));
 
        CURLcode curl_response = CURLE_FAILED_INIT;
        long status_code = 0;
        std::unique_ptr<ccf::curl::CurlRequest> request;
        ccf::curl::CurlRequest::ResponseCallback response_callback =
          [&curl_response, &status_code, &request](
            std::unique_ptr<ccf::curl::CurlRequest>&& request_,
            CURLcode curl_response_,
            long status_code_) {
            curl_response = curl_response_;
            status_code = status_code_;
            request = std::move(request_);
          };
 
        ccf::curl::CurlRequest::synchronous_perform(
          std::make_unique<ccf::curl::CurlRequest>(
            std::move(curl_easy),
            HTTP_GET,
            snapshot_url,
            std::move(headers),
            nullptr, // No request body
            std::move(response_body),
            std::move(response_callback)));
 
        if (curl_response != CURLE_OK)
        {
          throw std::runtime_error(fmt::format(
            "Error fetching snapshot redirect from {}: {} ({})",
            request->get_url(),
            curl_easy_strerror(curl_response),
            status_code));
        }
 
        if (status_code == HTTP_STATUS_NOT_FOUND)
        {
          LOG_INFO_FMT("Peer has no suitable snapshot");
          return std::nullopt;
        }
 
        if (status_code == HTTP_STATUS_PARTIAL_CONTENT)
        {
          process_partial_response(*request);
 
          response_body = std::move(request->get_response_ptr());
          curl_easy = std::move(request->get_easy_handle_ptr());
          break;
        }
 
        EXPECT_HTTP_RESPONSE_STATUS(
          request,
          status_code,
          HTTP_STATUS_PERMANENT_REDIRECT,
          request->get_response_ptr());
 
        char* redirect_url = nullptr;
        CHECK_CURL_EASY_GETINFO(
          request->get_easy_handle(), CURLINFO_REDIRECT_URL, &redirect_url);
        if (redirect_url == nullptr)
        {
          throw std::runtime_error(
            "Redirect response found, but CURLINFO_REDIRECT_URL returned no "
            "value");
        }
 
        LOG_DEBUG_FMT(
          "Snapshot fetch received redirect response with location {}",
          redirect_url);
        snapshot_url = redirect_url;
 
        response_body = std::move(request->get_response_ptr());
        curl_easy = std::move(request->get_easy_handle_ptr());
 
        // Ignore any body from redirect responses
        response_body->buffer.clear();
      }
 
      while (!fetched_all)
      {
        ccf::curl::UniqueSlist headers;
        headers.append(
          ccf::http::headers::RANGE,
          fmt::format("bytes={}-{}", range_start, inclusive_range_end));
 
        std::unique_ptr<ccf::curl::CurlRequest> snapshot_range_request;
        CURLcode curl_response = CURLE_OK;
        long snapshot_range_status_code = 0;
 
        ccf::curl::CurlRequest::ResponseCallback snapshot_response_callback =
          [&](
            std::unique_ptr<ccf::curl::CurlRequest>&& request_,
            CURLcode curl_response_,
            long status_code_) {
            snapshot_range_request = std::move(request_);
            curl_response = curl_response_;
            snapshot_range_status_code = status_code_;
          };
 
        ccf::curl::CurlRequest::synchronous_perform(
          std::make_unique<ccf::curl::CurlRequest>(
            std::move(curl_easy),
            HTTP_GET,
            snapshot_url,
            std::move(headers),
            nullptr, // No request body
            std::move(response_body),
            snapshot_response_callback));
        if (curl_response != CURLE_OK)
        {
          throw std::runtime_error(fmt::format(
            "Error fetching snapshot chunk range from {}: {} ({})",
            snapshot_range_request->get_url(),
            curl_easy_strerror(curl_response),
            snapshot_range_status_code));
        }
        EXPECT_HTTP_RESPONSE_STATUS(
          snapshot_range_request,
          snapshot_range_status_code,
          HTTP_STATUS_PARTIAL_CONTENT,
          snapshot_range_request->get_response_ptr());
 
        process_partial_response(*snapshot_range_request);
 
        response_body = std::move(snapshot_range_request->get_response_ptr());
        curl_easy = std::move(snapshot_range_request->get_easy_handle_ptr());
      }
 
      const auto url_components = ccf::nonstd::split(snapshot_url, "/");
      const std::string snapshot_name(url_components.back());
 
      return SnapshotResponse{snapshot_name, std::move(response_body->buffer)};
    }
    catch (const std::exception& e)
    {
      LOG_FAIL_FMT("Error during snapshot fetch: {}", e.what());
      return std::nullopt;
    }
  }
 
  static std::optional<SnapshotResponse> fetch_from_peer(
    const std::string& peer_address,
    const std::vector<uint8_t>& peer_ca,
    size_t max_attempts,
    size_t retry_delay_ms,
    size_t max_size,
    std::optional<size_t> since_seqno = std::nullopt)
  {
    for (size_t attempt = 0; attempt < max_attempts; ++attempt)
    {
      LOG_INFO_FMT(
        "Fetching snapshot from {} since {} (attempt {}/{})",
        peer_address,
        since_seqno.has_value() ? std::to_string(*since_seqno) : "any",
        attempt + 1,
        max_attempts);
 
      if (attempt > 0)
      {
        std::this_thread::sleep_for(std::chrono::milliseconds(retry_delay_ms));
      }
 
      auto response =
        try_fetch_from_peer(peer_address, peer_ca, max_size, since_seqno);
      if (response.has_value())
      {
        return response;
      }
    }
    LOG_INFO_FMT(
      "Exceeded maximum snapshot fetch retries ({}), giving up", max_attempts);
    return std::nullopt;
  }
}