code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#pragma once

#include <rotor.hpp>
#include "model/cluster.h"
#include "config/bep.h"
#include "messages.h"
#include "utils/log.h"
#include "transport/stream.h"

namespace syncspirit::net {

namespace r = rotor;

struct initiator_actor_config_t : public r::actor_config_t {
    model::device_id_t peer_device_id;
    utils::uri_container_t uris;
    utils::bytes_t relay_session;
    const utils::key_pair_t *ssl_pair;
    std::optional<tcp_socket_t> sock;
    model::cluster_ptr_t cluster;
    r::address_ptr_t sink;
    r::message_ptr_t custom;
    r::supervisor_t *router;
    std::string_view alpn;
    bool relay_enabled;
};

template <typename Actor> struct initiator_actor_config_builder_t : r::actor_config_builder_t<Actor> {
    using builder_t = typename Actor::template config_builder_t<Actor>;
    using parent_t = r::actor_config_builder_t<Actor>;
    using parent_t::parent_t;

    builder_t &&peer_device_id(const model::device_id_t &value) && noexcept {
        parent_t::config.peer_device_id = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&uris(const utils::uri_container_t &value) && noexcept {
        parent_t::config.uris = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&relay_session(utils::bytes_view_t value) && noexcept {
        parent_t::config.relay_session = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&ssl_pair(const utils::key_pair_t *value) && noexcept {
        parent_t::config.ssl_pair = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&sock(tcp_socket_t value) && noexcept {
        parent_t::config.sock = std::move(value);
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&cluster(const model::cluster_ptr_t &value) && noexcept {
        parent_t::config.cluster = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&sink(const r::address_ptr_t &value) && noexcept {
        parent_t::config.sink = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&custom(r::message_ptr_t value) && noexcept {
        parent_t::config.custom = std::move(value);
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&router(r::supervisor_t &value) && noexcept {
        parent_t::config.router = &value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&alpn(std::string_view value) && noexcept {
        parent_t::config.alpn = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&relay_enabled(bool value) && noexcept {
        parent_t::config.relay_enabled = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API initiator_actor_t : r::actor_base_t {
    using config_t = initiator_actor_config_t;
    template <typename Actor> using config_builder_t = initiator_actor_config_builder_t<Actor>;

    initiator_actor_t(config_t &config);
    void configure(r::plugin::plugin_base_t &plugin) noexcept override;
    void on_start() noexcept override;
    void shutdown_start() noexcept override;
    void shutdown_finish() noexcept override;

    template <typename T> auto &access() noexcept;

  private:
    using resolve_it_t = payload::address_response_t::resolve_results_t::iterator;
    enum class role_t { active, passive, relay_passive };

    void initiate_passive() noexcept;
    void initiate_active() noexcept;
    void initiate_relay_passive() noexcept;
    void initiate_active_tls(const utils::uri_ptr_t &uri) noexcept;
    void initiate_active_relay(const utils::uri_ptr_t &uri) noexcept;
    void initiate_handshake() noexcept;
    void join_session() noexcept;
    void request_relay_connection() noexcept;
    void resolve(const utils::uri_ptr_t &uri) noexcept;

    void on_resolve(message::resolve_response_t &res) noexcept;
    void on_connect(const tcp::endpoint &) noexcept;
    void on_io_error(const sys::error_code &ec, r::plugin::resource_id_t resource) noexcept;
    void on_handshake(bool valid_peer, utils::x509_t &peer_cert, const tcp::endpoint &peer_endpoint,
                      const model::device_id_t *peer_device) noexcept;
    void on_read_relay(size_t bytes) noexcept;
    void on_read_relay_active(size_t bytes) noexcept;
    void on_write(size_t bytes) noexcept;

    model::device_id_t peer_device_id;
    utils::uri_container_t uris;
    utils::bytes_t relay_rx;
    utils::bytes_t relay_tx;
    utils::bytes_t relay_key;
    const utils::key_pair_t &ssl_pair;
    std::optional<tcp_socket_t> sock;
    model::cluster_ptr_t cluster;
    r::address_ptr_t sink;
    r::message_ptr_t custom;
    r::supervisor_t &router;
    std::string_view alpn;

    utils::uri_ptr_t active_uri;
    transport::stream_sp_t transport;
    r::address_ptr_t resolver;
    r::address_ptr_t coordinator;
    size_t uri_idx = 0;
    utils::logger_t log;
    tcp::endpoint remote_endpoint;
    bool connected = false;
    role_t role = role_t::passive;
    utils::bytes_t rx_buff;
    bool success = false;
    bool relaying = false;
};

namespace payload {

struct peer_connected_t {
    transport::stream_sp_t transport;
    model::device_id_t peer_device_id;
    tcp::endpoint remote_endpoint;
    std::string proto;
    r::message_ptr_t custom;
};

} // namespace payload

namespace message {

using peer_connected_t = r::message_t<payload::peer_connected_t>;

}

} // namespace syncspirit::net


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "local_discovery_actor.h"
#include "names.h"
#include "proto/bep_support.h"
#include "utils/error_code.h"
#include "utils/format.hpp"
#include "utils/time.h"
#include "model/diff/contact/update_contact.h"
#include "model/diff/contact/ignored_connected.h"
#include "model/diff/contact/unknown_connected.h"
#include "model/diff/modify/add_pending_device.h"
#include "model/messages.h"
#include <fmt/ranges.h>

using namespace syncspirit;
using namespace syncspirit::net;

static const constexpr std::size_t BUFF_SZ = 1500;

namespace {
namespace resource {
r::plugin::resource_id_t send = 0;
r::plugin::resource_id_t read = 1;
r::plugin::resource_id_t timer = 2;
} // namespace resource
} // namespace

local_discovery_actor_t::local_discovery_actor_t(config_t &cfg)
    : r::actor_base_t{cfg}, frequency{r::pt::milliseconds(cfg.frequency)},
      strand{static_cast<ra::supervisor_asio_t *>(cfg.supervisor)->get_strand()}, broadcast_sock{strand.context()},
      sock{strand.context()}, port{cfg.port}, cluster{cfg.cluster} {
    rx_buff.resize(BUFF_SZ);
    tx_buff.resize(BUFF_SZ);
}

void local_discovery_actor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    r::actor_base_t::configure(plugin);
    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        p.set_identity("net.local_discovery", false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>(
        [&](auto &p) { p.discover_name(names::coordinator, coordinator, true).link(false); });
}

void local_discovery_actor_t::init() noexcept {
    LOG_TRACE(log, "init, will announce to port = {}", port);

    sys::error_code ec;

    auto bc_endpoint = udp::endpoint(asio::ip::address_v4::any(), port);
    broadcast_sock.open(bc_endpoint.protocol(), ec);
    if (ec) {
        LOG_WARN(log, "init, can't open broadcast socket :: {}", ec.message());
        return do_shutdown(make_error(ec));
    }

    broadcast_sock.set_option(udp_socket_t::broadcast(true), ec);
    if (ec) {
        LOG_WARN(log, "init, can't set broadcast option :: {}", ec.message());
        return do_shutdown(make_error(ec));
    }

    auto listen_endpoint = udp::endpoint{asio::ip::address_v4::any(), port};
    sock.open(listen_endpoint.protocol(), ec);
    if (ec) {
        LOG_WARN(log, "init, can't open socket :: {}", ec.message());
        return do_shutdown(make_error(ec));
    }

    sock.bind(listen_endpoint, ec);
    if (ec) {
        LOG_WARN(log, "init, can't bind socket {} :: {}", listen_endpoint, ec.message());
        return do_shutdown(make_error(ec));
    }
}

void local_discovery_actor_t::on_start() noexcept {
    LOG_TRACE(log, "on_start");
    r::actor_base_t::on_start();
    init();
    do_read();
    announce();
}

void local_discovery_actor_t::shutdown_start() noexcept {
    if (resources->has(resource::timer)) {
        cancel_timer(*timer_request);
    }
    sys::error_code ec;
    if (resources->has(resource::send)) {
        broadcast_sock.cancel(ec);
        if (ec) {
            LOG_WARN(log, "shutdown_start, socket cancellation error:: {}", ec.message());
        }
    }
    if (resources->has(resource::read)) {
        sock.cancel(ec);
        if (ec) {
            LOG_WARN(log, "shutdown_start, socket cancellation error:: {}", ec.message());
        }
    }
    r::actor_base_t::shutdown_start();
}

void local_discovery_actor_t::announce() noexcept {
    auto instance = std::uint64_t{reinterpret_cast<std::uint64_t>(this)};

    auto device = cluster->get_device();
    auto &uris = device->get_uris();
    if (!uris.empty()) {
        auto digest = device->device_id().get_sha256();
        auto sz = proto::make_announce_message(tx_buff, digest, uris, instance);
        auto buff = asio::const_buffer(tx_buff.data(), sz);
        auto fwd_send =
            ra::forwarder_t(*this, &local_discovery_actor_t::on_write, &local_discovery_actor_t::on_write_error);
        auto bc_endpoint = udp::endpoint(asio::ip::address_v4::broadcast(), port);
        broadcast_sock.async_send_to(buff, bc_endpoint, std::move(fwd_send));
        resources->acquire(resource::send);
        LOG_TRACE(log, "announce has been sent (to = {}, instance = {:x})", bc_endpoint, instance);
    } else {
        LOG_TRACE(log, "announce() skipping");
    }

    timer_request = start_timer(frequency, *this, &local_discovery_actor_t::on_timer);
    resources->acquire(resource::timer);
}

void local_discovery_actor_t::do_read() noexcept {
    auto buff = asio::buffer(rx_buff.data(), BUFF_SZ);
    auto fwd_receive =
        ra::forwarder_t(*this, &local_discovery_actor_t::on_read, &local_discovery_actor_t::on_read_error);
    sock.async_receive_from(buff, peer_endpoint, std::move(fwd_receive));
    resources->acquire(resource::read);
}

void local_discovery_actor_t::on_timer(r::request_id_t, bool cancelled) noexcept {
    resources->release(resource::timer);
    timer_request.reset();

    if (!cancelled) {
        announce();
    }
}

void local_discovery_actor_t::on_read(size_t bytes) noexcept {
    resources->release(resource::read);
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    // LOG_TRACE(log, "local_discovery_actor_t::on_read");
    auto view = utils::bytes_view_t(rx_buff.data(), bytes);
    auto result = proto::parse_announce(view);
    if (!result) {
        LOG_TRACE(log, "on_read, cannot parse incoming UDP packet {} bytes from {} :: {}", bytes, peer_endpoint,
                  result.error().message());
    } else {
        auto &msg = result.value();
        auto sha = proto::get_id(msg);
        auto device_id = model::device_id_t::from_sha256(sha);
        if (device_id) {
            utils::uri_container_t uris;
            auto addresses_count = proto::get_addresses_size(msg);
            for (int i = 0; i < addresses_count; ++i) {
                auto address = proto::get_addresses(msg, i);
                auto uri = utils::parse(address);
                if (uri && uri->has_port()) {
                    uris.emplace_back(std::move(uri));
                }
            }
            if (!uris.empty()) {
                handle(*device_id, uris);
            } else {
                LOG_WARN(log, "on_read, no valid uris from: {} / {}", peer_endpoint, *device_id);
            }
        } else {
            LOG_WARN(log, "on_read, wrong device id, coming from: {}", peer_endpoint);
        }
    }
    return do_read();
}

void local_discovery_actor_t::on_read_error(const sys::error_code &ec) noexcept {
    resources->release(resource::read);
    if (ec != asio::error::operation_aborted) {
        LOG_ERROR(log, "on_read_error, error = {}", ec.message());
        do_shutdown(make_error(ec));
    }
}

void local_discovery_actor_t::on_write(size_t) noexcept { resources->release(resource::send); }

void local_discovery_actor_t::on_write_error(const sys::error_code &ec) noexcept {
    resources->release(resource::send);
    if (ec != asio::error::operation_aborted) {
        LOG_ERROR(log, "on_write_error, error = {}", ec.message());
        do_shutdown(make_error(ec));
    }
}

struct filler_t {
    template <typename T> static auto fill(T &peer, const std::string &uris_str) -> db::SomeDevice {
        db::SomeDevice db;
        peer->serialize(db);
        auto last_seen = utils::as_seconds(pt::microsec_clock::local_time());
        db::set_address(db, uris_str);
        db::set_last_seen(db, last_seen);
        return db;
    }
};

void local_discovery_actor_t::handle(const model::device_id_t &device_id, utils::uri_container_t &uris) noexcept {
    if (device_id == cluster->get_device()->device_id()) {
        LOG_TRACE(log, "skipping self discovery");
        return;
    }
    if (uris.empty()) {
        LOG_TRACE(log, "no valid urls found for {}", device_id);
        return;
    }

    using namespace model::diff;
    auto diff = model::diff::cluster_diff_ptr_t{};
    auto uris_str = fmt::format("{}", fmt::join(uris, ", "));
    LOG_TRACE(log, "on_read, peer is available via {}", uris_str);

    if (auto peer = cluster->get_devices().by_sha256(device_id.get_sha256()); peer) {
        LOG_DEBUG(log, "device '{}' contacted", device_id);
        diff = new contact::update_contact_t(*cluster, device_id, uris);
    } else if (auto peer = cluster->get_ignored_devices().by_sha256(device_id.get_sha256()); peer) {
        LOG_DEBUG(log, "ignored device '{}' contacted", device_id);
        auto db = filler_t::fill(peer, uris_str);
        diff = new contact::ignored_connected_t(*cluster, device_id, std::move(db));
    } else if (auto peer = cluster->get_pending_devices().by_sha256(device_id.get_sha256()); peer) {
        auto db = filler_t::fill(peer, uris_str);
        diff = new contact::unknown_connected_t(*cluster, device_id, std::move(db));
    } else {
        db::SomeDevice db;
        auto last_seen = utils::as_seconds(pt::microsec_clock::local_time());
        db::set_name(db, device_id.get_short());
        db::set_address(db, uris_str);
        db::set_last_seen(db, last_seen);
        diff = new model::diff::modify::add_pending_device_t(device_id, db);
        diff->assign_sibling(new contact::unknown_connected_t(*cluster, device_id, std::move(db)));
    }
    if (diff) {
        send<model::payload::model_update_t>(coordinator, std::move(diff), this);
    }
}

repo_name: basiliscos/syncspirit

license: mit


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#pragma once

#include "messages.h"
#include "model/cluster.h"
#include "utils/log.h"
#include <boost/asio.hpp>
#include <optional>

namespace syncspirit {
namespace net {

namespace outcome = boost::outcome_v2;

struct local_discovery_actor_config_t : r::actor_config_t {
    std::uint16_t port;
    std::uint32_t frequency;
    model::cluster_ptr_t cluster;
};

template <typename Actor> struct local_discovery_actor_config_builder_t : r::actor_config_builder_t<Actor> {
    using builder_t = typename Actor::template config_builder_t<Actor>;
    using parent_t = r::actor_config_builder_t<Actor>;
    using parent_t::parent_t;

    builder_t &&port(const std::uint16_t value) && noexcept {
        parent_t::config.port = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&frequency(const std::uint32_t value) && noexcept {
        parent_t::config.frequency = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&cluster(const model::cluster_ptr_t &value) && noexcept {
        parent_t::config.cluster = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API local_discovery_actor_t final : public r::actor_base_t {
    using config_t = local_discovery_actor_config_t;
    template <typename Actor> using config_builder_t = local_discovery_actor_config_builder_t<Actor>;

    explicit local_discovery_actor_t(config_t &cfg);

    void on_start() noexcept override;
    void shutdown_start() noexcept override;
    void configure(r::plugin::plugin_base_t &plugin) noexcept override;

  private:
    void init() noexcept;
    void announce() noexcept;
    void do_read() noexcept;

    void on_read(size_t bytes) noexcept;
    void on_read_error(const sys::error_code &ec) noexcept;
    void on_write(size_t bytes) noexcept;
    void on_write_error(const sys::error_code &ec) noexcept;
    void on_timer(r::request_id_t, bool cancelled) noexcept;

    void handle(const model::device_id_t &device_id, utils::uri_container_t &uris) noexcept;

    utils::logger_t log;
    r::pt::time_duration frequency;
    asio::io_context::strand &strand;
    udp_socket_t broadcast_sock;
    udp_socket_t sock;
    r::address_ptr_t coordinator;
    utils::bytes_t rx_buff;
    utils::bytes_t tx_buff;
    std::uint16_t port;
    udp::endpoint peer_endpoint;
    std::optional<r::request_id_t> timer_request;
    std::optional<r::request_id_t> endpoint_request;
    model::cluster_ptr_t cluster;
};

} // namespace net
} // namespace syncspirit


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "messages.h"
#include "model/cluster.h"

using namespace syncspirit::net::payload;

block_request_t::block_request_t(const model::file_info_ptr_t &file_, size_t block_index_) noexcept {
    auto fi = file_->get_folder_info();
    folder_id = fi->get_folder()->get_id();
    file_name = file_->get_name();
    sequence = file_->get_sequence();
    block_index = block_index_;
    auto block = file_->get_blocks()[block_index_].get();
    block_hash = block->get_hash();
    block_offset = block->get_file_blocks().front().get_offset();
    block_size = block->get_size();
}

auto block_request_t::get_block(model::cluster_t &cluster, model::device_t &peer) noexcept -> model::file_block_t {
    auto folder = cluster.get_folders().by_id(folder_id);
    if (!folder || folder->is_suspended()) {
        return {};
    }
    auto fi = folder->get_folder_infos().by_device(peer);
    if (!fi) {
        return {};
    }
    auto file = fi->get_file_infos().by_name(file_name);
    if (!file) {
        return {};
    }
    if (file->get_sequence() != sequence) {
        return {};
    }
    auto &blocks = file->get_blocks();
    if (block_index >= blocks.size()) {
        return {};
    }
    auto &file_blocks = blocks[block_index]->get_file_blocks();
    for (auto &fb : file_blocks) {
        if (fb.file() == file) {
            return fb;
        }
    }
    return {};
}


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#pragma once

#include <boost/asio.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <rotor/asio.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>

#include <memory>
#include <optional>

#include "model/diff/cluster_diff.h"
#include "model/file_info.h"
#include "transport/base.h"
#include "utils/bytes.h"
#include "utils/dns.h"

namespace syncspirit {
namespace net {

namespace r = rotor;
namespace ra = rotor::asio;
namespace asio = boost::asio;
namespace pt = boost::posix_time;
namespace http = boost::beast::http;
namespace sys = boost::system;
namespace ssl = asio::ssl;

using tcp = asio::ip::tcp;
using udp = asio::ip::udp;
using v4 = asio::ip::address_v4;
using udp_socket_t = udp::socket;
using tcp_socket_t = tcp::socket;

namespace payload {

using cluster_config_ptr_t = std::unique_ptr<proto::ClusterConfig>;

struct address_response_t : public r::arc_base_t<address_response_t> {
    using resolve_results_t = std::vector<asio::ip::address>;

    explicit address_response_t(resolve_results_t results_) : results{results_} {};
    resolve_results_t results;
};

struct address_request_t : r::arc_base_t<address_request_t>, utils::dns_query_t {
    using response_t = r::intrusive_ptr_t<address_response_t>;
    address_request_t(std::string_view host_, std::uint16_t port_ = 0)
        : utils::dns_query_t{std::string(host_)}, port{port_} {}

    std::uint16_t port;
};

struct http_response_t : public r::arc_base_t<http_response_t> {
    using raw_http_response_t = http::response<http::string_body>;
    raw_http_response_t response;
    std::size_t bytes;
    std::optional<asio::ip::address> local_addr;

    http_response_t(raw_http_response_t &&response_, std::size_t bytes_,
                    std::optional<asio::ip::address> local_addr_ = {})
        : response(std::move(response_)), bytes{bytes_}, local_addr{std::move(local_addr_)} {}
};

struct http_request_t : r::arc_base_t<http_request_t> {
    using rx_buff_t = boost::beast::flat_buffer;
    using rx_buff_ptr_t = std::shared_ptr<rx_buff_t>;
    using duration_t = r::pt::time_duration;
    using ssl_option_t = transport::ssl_option_t;
    using response_t = r::intrusive_ptr_t<http_response_t>;

    utils::uri_ptr_t url;
    utils::bytes_t data;
    rx_buff_ptr_t rx_buff;
    std::size_t rx_buff_size;
    ssl_option_t ssl_context;
    bool local_ip = false;
    bool debug = false;
    r::message_ptr_t custom;

    template <typename URI>
    http_request_t(URI &&url_, utils::bytes_t &&data_, rx_buff_ptr_t rx_buff_, std::size_t rx_buff_size_,
                   bool local_ip_, bool debug_ = false, const r::message_ptr_t &custom_ = {})
        : url{std::forward<URI>(url_)}, data{std::move(data_)}, rx_buff{rx_buff_}, rx_buff_size{rx_buff_size_},
          local_ip{local_ip_}, debug{debug_}, custom{custom_} {}

    template <typename URI>
    http_request_t(URI &&url_, utils::bytes_t &&data_, rx_buff_ptr_t rx_buff_, std::size_t rx_buff_size_,
                   transport::ssl_junction_t &&ssl_, bool debug = false, const r::message_ptr_t &custom_ = {})
        : http_request_t(std::forward<URI>(url_), std::move(data_), rx_buff_, rx_buff_size_, {}, debug, custom_) {
        ssl_context = ssl_option_t(std::move(ssl_));
    }
};

struct http_close_connection_t {};

struct announce_notification_t {
    r::address_ptr_t source;
};

struct load_cluster_response_t {
    model::diff::cluster_diff_ptr_t diff;
};

struct load_cluster_request_t {
    using response_t = load_cluster_response_t;
};

struct start_reading_t {
    r::address_ptr_t controller;
    bool start;
};

struct termination_t {
    r::extended_error_ptr_t ee;
};

using forwarded_message_t =
    std::variant<proto::ClusterConfig, proto::Index, proto::IndexUpdate, proto::Request, proto::DownloadProgress>;

struct block_response_t {
    utils::bytes_t data;
};

struct block_request_t {
    using response_t = block_response_t;
    std::string folder_id;
    std::string file_name;
    std::int64_t sequence;
    size_t block_index;
    std::int64_t block_offset;
    std::uint32_t block_size;
    utils::bytes_t block_hash;
    block_request_t(const model::file_info_ptr_t &file, size_t block_index) noexcept;

    model::file_block_t get_block(model::cluster_t &, model::device_t &peer) noexcept;
};

struct connect_response_t {
    transport::stream_sp_t transport;
    tcp::endpoint remote_endpoint;
};

struct connect_request_t {
    using response_t = connect_response_t;
    model::device_id_t device_id;
    utils::uri_ptr_t uri;
    std::string_view alpn;
};

struct transfer_data_t {
    utils::bytes_t data;
};

struct transfer_push_t {
    uint32_t bytes;
};

struct transfer_pop_t {
    uint32_t bytes;
};

struct db_info_response_t {
    uint32_t page_size = 0;
    uint32_t tree_depth = 0;
    uint64_t leaf_pages = 0;
    uint64_t overflow_pages = 0;
    uint64_t ms_branch_pages = 0;
    uint64_t entries = 0;
};

struct db_info_request_t {
    using response_t = db_info_response_t;
};

} // end of namespace payload

namespace message {

using announce_notification_t = r::message_t<payload::announce_notification_t>;

using resolve_request_t = r::request_traits_t<payload::address_request_t>::request::message_t;
using resolve_response_t = r::request_traits_t<payload::address_request_t>::response::message_t;
using resolve_cancel_t = r::request_traits_t<payload::address_request_t>::cancel::message_t;

using http_request_t = r::request_traits_t<payload::http_request_t>::request::message_t;
using http_response_t = r::request_traits_t<payload::http_request_t>::response::message_t;
using http_cancel_t = r::request_traits_t<payload::http_request_t>::cancel::message_t;
using http_close_connection_t = r::message_t<payload::http_close_connection_t>;

using load_cluster_request_t = r::request_traits_t<payload::load_cluster_request_t>::request::message_t;
using load_cluster_response_t = r::request_traits_t<payload::load_cluster_request_t>::response::message_t;

using start_reading_t = r::message_t<payload::start_reading_t>;
using forwarded_message_t = r::message_t<payload::forwarded_message_t>;
using termination_signal_t = r::message_t<payload::termination_t>;
using transfer_data_t = r::message_t<payload::transfer_data_t>;
using transfer_push_t = r::message_t<payload::transfer_push_t>;
using transfer_pop_t = r::message_t<payload::transfer_pop_t>;

using block_request_t = r::request_traits_t<payload::block_request_t>::request::message_t;
using block_response_t = r::request_traits_t<payload::block_request_t>::response::message_t;

using connect_request_t = r::request_traits_t<payload::connect_request_t>::request::message_t;
using connect_response_t = r::request_traits_t<payload::connect_request_t>::response::message_t;

using db_info_request_t = r::request_traits_t<payload::db_info_request_t>::request::message_t;
using db_info_response_t = r::request_traits_t<payload::db_info_request_t>::response::message_t;

} // end of namespace message

} // namespace net
} // namespace syncspirit


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "names.h"

using namespace syncspirit::net;

const char *names::peer_supervisor = "net.peer_supervisor";
const char *names::coordinator = "net.coordinator";
const char *names::sink = "net.sink";
const char *names::resolver = "net.resolver";
const char *names::http10 = "net.http10";
const char *names::http11_gda = "net.http11_gda";
const char *names::http11_relay = "net.http11_relay";
const char *names::hasher_proxy = "net.hasher_proxy";
const char *names::fs_actor = "fs.actor";
const char *names::fs_scanner = "fs.scanner";
const char *names::fs_scheduler = "fs.scheduler";


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#pragma once

#include "syncspirit-export.h"

namespace syncspirit {
namespace net {

struct SYNCSPIRIT_API names {
    static const char *peer_supervisor;
    static const char *coordinator;
    static const char *sink;
    static const char *resolver;
    static const char *http10;
    static const char *http11_gda;
    static const char *http11_relay;
    static const char *hasher_proxy;
    static const char *fs_actor;
    static const char *fs_scanner;
    static const char *fs_scheduler;
};

} // namespace net
}; // namespace syncspirit


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "config/utils.h"
#include "net_supervisor.h"
#include "global_discovery_actor.h"
#include "local_discovery_actor.h"
#include "cluster_supervisor.h"
#include "acceptor_actor.h"
#include "ssdp_actor.h"
#include "http_actor.h"
#include "resolver_actor.h"
#include "peer_supervisor.h"
#include "dialer_actor.h"
#include "db_actor.h"
#include "relay_actor.h"
#include "sink_actor.h"
#include "names.h"
#include "model/diff/load/load_cluster.h"
#include <filesystem>
#include <ctime>

namespace bfs = std::filesystem;
using namespace syncspirit::net;

net_supervisor_t::net_supervisor_t(net_supervisor_t::config_t &cfg)
    : parent_t{cfg}, sequencer{cfg.sequencer}, app_config{cfg.app_config}, cluster_copies{cfg.cluster_copies} {
    auto log = utils::get_logger(names::coordinator);
    auto &files_cfg = app_config.global_announce_config;
    auto result = utils::load_pair(files_cfg.cert_file.c_str(), files_cfg.key_file.c_str());
    if (!result) {
        LOG_CRITICAL(log, "cannot load certificate/key pair :: {}", result.error().message());
        throw result.error();
    }
    ssl_pair = std::move(result.value());
    auto device_id_opt = model::device_id_t::from_cert(ssl_pair.cert_data);
    if (!device_id_opt) {
        LOG_CRITICAL(log, "cannot create device_id from certificate");
        throw "cannot create device_id from certificate";
    }
    auto &device_id = device_id_opt.value();
    LOG_INFO(log, "{}, device name = {}, device id = {}", names::coordinator, app_config.device_name,
             device_id.get_value());

    auto cn = utils::get_common_name(ssl_pair.cert.get());
    if (!cn) {
        LOG_CRITICAL(log, "cannot get common name from certificate");
        throw "cannot get common name from certificate";
    }
    auto device_opt = model::device_t::create(device_id, app_config.device_name, cn.value());
    if (!device_opt) {
        LOG_CRITICAL(log, "cannot get common name from certificate");
        throw "cannot get common name from certificate";
    }

    auto device = model::device_ptr_t();
    device = new model::local_device_t(device_id, app_config.device_name, cn.value());
    auto simultaneous_writes = app_config.bep_config.blocks_simultaneous_write;
    cluster = new model::cluster_t(device, static_cast<int32_t>(simultaneous_writes));

    auto &gcfg = app_config.global_announce_config;
    if (gcfg.enabled) {
        auto global_device_opt = model::device_id_t::from_string(gcfg.device_id);
        if (!global_device_opt) {
            LOG_CRITICAL(log, "invalid global device id :: {}", gcfg.device_id);
            throw "invalid global device id";
        }
        global_device = std::move(global_device_opt.value());
    }
}

void net_supervisor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    parent_t::configure(plugin);
    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        p.set_identity(names::coordinator, false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>(
        [&](auto &p) { p.register_name(names::coordinator, get_address()); });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        p.subscribe_actor(&net_supervisor_t::on_model_update);
        p.subscribe_actor(&net_supervisor_t::on_load_cluster);
        p.subscribe_actor(&net_supervisor_t::on_model_request);
        launch_early();
    });
}

void net_supervisor_t::on_child_shutdown(actor_base_t *actor) noexcept {
    parent_t::on_child_shutdown(actor);
    auto &reason = actor->get_shutdown_reason();
    LOG_TRACE(log, "on_child_shutdown, '{}' due to {} ", actor->get_identity(), reason->message());
}

void net_supervisor_t::on_child_init(actor_base_t *actor, const r::extended_error_ptr_t &ec) noexcept {
    parent_t::on_child_init(actor, ec);
    auto &child_addr = actor->get_address();
    if (!ec && db_addr && child_addr == db_addr) {
        LOG_TRACE(log, "on_child_init, db has been launched, let's load it...");
        load_db();
    }
}

void net_supervisor_t::shutdown_finish() noexcept {
    db_addr.reset();
    ra::supervisor_asio_t::shutdown_finish();
}

void net_supervisor_t::launch_early() noexcept {
    auto timeout = shutdown_timeout * 9 / 10;
    db_addr = create_actor<db_actor_t>()
                  .timeout(timeout)
                  .db_dir(app_config.config_path / "mdbx-db")
                  .db_config(app_config.db_config)
                  .cluster(cluster)
                  .escalate_failure()
                  .finish()
                  ->get_address();
    create_actor<sink_actor_t>().timeout(timeout).escalate_failure().finish();
}

void net_supervisor_t::load_db() noexcept {
    auto timeout = init_timeout * 9 / 10;
    request<payload::load_cluster_request_t>(db_addr).send(timeout);
}

void net_supervisor_t::seed_model() noexcept {
    auto message =
        r::make_routed_message<model::payload::model_update_t>(address, db_addr, std::move(load_diff), nullptr);
    put(std::move(message));
}

void net_supervisor_t::on_load_cluster(message::load_cluster_response_t &message) noexcept {
    auto &ee = message.payload.ee;
    if (ee) {
        LOG_ERROR(log, "cannot load cluster : {}", ee->message());
        return do_shutdown(ee);
    }
    LOG_TRACE(log, "on_load_cluster");
    auto &res = message.payload.res;

    load_diff = std::move(res.diff);
    if (cluster_copies == 0) {
        seed_model();
    }
}

void net_supervisor_t::on_model_request(model::message::model_request_t &message) noexcept {
    --cluster_copies;
    LOG_TRACE(log, "on_cluster_seed, left = {}", cluster_copies);
    auto my_device = cluster->get_device();
    auto device = model::device_ptr_t();
    device = new model::local_device_t(my_device->device_id(), app_config.device_name, "");
    auto simultaneous_writes = app_config.bep_config.blocks_simultaneous_write;
    auto cluster_copy = new model::cluster_t(device, static_cast<int32_t>(simultaneous_writes));
    reply_to(message, std::move(cluster_copy));
    if (cluster_copies == 0 && load_diff) {
        seed_model();
    }
}

void net_supervisor_t::on_model_update(model::message::model_update_t &message) noexcept {
    LOG_TRACE(log, "on_model_update");
    auto &diff = *message.payload.diff;
    auto r = diff.apply(*cluster, *this);
    if (!r) {
        LOG_ERROR(log, "error applying model diff: {}", r.assume_error().message());
        auto ee = make_error(r.assume_error());
        do_shutdown(ee);
    }
    r = diff.visit(*this, nullptr);
    if (!r) {
        auto ee = make_error(r.assume_error());
        do_shutdown(ee);
    }
}

auto net_supervisor_t::operator()(const model::diff::load::load_cluster_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (!cluster->is_tainted()) {

        auto &ignored_devices = cluster->get_ignored_devices();
        auto &ignored_folders = cluster->get_ignored_folders();
        auto &pending_folders = cluster->get_pending_folders();
        auto &pending_devices = cluster->get_pending_devices();
        auto &devices = cluster->get_devices();
        auto &folders = cluster->get_folders();
        size_t files = 0;
        for (const auto &it : folders) {
            auto &folder_info = it.item;
            if (!folder_info) {
                continue;
            }
            auto fi = folder_info->get_folder_infos().by_device(*cluster->get_device());
            files += fi->get_file_infos().size();
        }
        auto pending_folders_sz = std::distance(pending_folders.begin(), pending_folders.end());
        LOG_DEBUG(log,
                  "load cluster, devices = {}, folders = {}, local files = {}, blocks = {}, ignored devices = {}, "
                  "ignored folders = {}, pending folders = {}, pending devices = {}",
                  devices.size(), folders.size(), files, cluster->get_blocks().size(), ignored_devices.size(),
                  ignored_folders.size(), pending_folders_sz, pending_devices.size());

        create_actor<cluster_supervisor_t>()
            .timeout(shutdown_timeout * 9 / 10)
            .strand(strand)
            .cluster(cluster)
            .sequencer(sequencer)
            .config(app_config)
            .escalate_failure()
            .finish();
        launch_net();
    }
    return diff.visit_next(*this, custom);
}

void net_supervisor_t::launch_net() noexcept {
    LOG_INFO(log, "launching network services");

    if (app_config.upnp_config.enabled) {
        auto factory = [this](r::supervisor_t &, const r::address_ptr_t &spawner) -> r::actor_ptr_t {
            auto timeout = shutdown_timeout * 9 / 10;
            return create_actor<ssdp_actor_t>()
                .timeout(timeout)
                .upnp_config(app_config.upnp_config)
                .cluster(cluster)
                .spawner_address(spawner)
                .finish();
        };
        spawn(factory).restart_period(pt::seconds{5}).restart_policy(r::restart_policy_t::fail_only).spawn();
    }

    if (app_config.local_announce_config.enabled) {
        auto factory = [this](r::supervisor_t &, const r::address_ptr_t &spawner) -> r::actor_ptr_t {
            auto timeout = shutdown_timeout * 9 / 10;
            auto &cfg = app_config.local_announce_config;
            return create_actor<local_discovery_actor_t>()
                .port(cfg.port)
                .frequency(cfg.frequency)
                .cluster(cluster)
                .timeout(timeout)
                .spawner_address(spawner)
                .finish();
        };
        spawn(factory).restart_period(pt::seconds{5}).restart_policy(r::restart_policy_t::fail_only).spawn();
    }

    if (app_config.global_announce_config.enabled) {
        auto timeout = shutdown_timeout * 9 / 10;
        auto io_timeout = shutdown_timeout * 8 / 10;
        create_actor<http_actor_t>()
            .timeout(timeout)
            .request_timeout(io_timeout)
            .resolve_timeout(io_timeout)
            .registry_name(names::http11_gda)
            .keep_alive(true)
            .escalate_failure()
            .finish();

        auto factory = [this](r::supervisor_t &, const r::address_ptr_t &spawner) -> r::actor_ptr_t {
            auto &gcfg = app_config.global_announce_config;
            auto timeout = shutdown_timeout * 9 / 10;
            return create_actor<global_discovery_actor_t>()
                .timeout(timeout)
                .cluster(cluster)
                .ssl_pair(&ssl_pair)
                .announce_url(gcfg.announce_url)
                .device_id(std::move(global_device))
                .rx_buff_size(gcfg.rx_buff_size)
                .io_timeout(gcfg.timeout)
                .debug(gcfg.debug)
                .spawner_address(spawner)
                .finish();
        };
        spawn(factory).restart_period(pt::seconds{5}).restart_policy(r::restart_policy_t::fail_only).spawn();
    }

    if (app_config.relay_config.enabled) {
        auto timeout = shutdown_timeout * 9 / 10;
        auto io_timeout = shutdown_timeout * 8 / 10;
        create_actor<http_actor_t>()
            .timeout(timeout)
            .request_timeout(io_timeout)
            .resolve_timeout(io_timeout)
            .registry_name(names::http11_relay)
            .keep_alive(true)
            .escalate_failure()
            .finish();

        auto factory = [this](r::supervisor_t &, const r::address_ptr_t &spawner) -> r::actor_ptr_t {
            auto timeout = shutdown_timeout * 9 / 10;
            return create_actor<relay_actor_t>()
                .timeout(timeout)
                .relay_config(app_config.relay_config)
                .cluster(cluster)
                .spawner_address(spawner)
                .finish();
        };
        spawn(factory).restart_period(pt::seconds{5}).restart_policy(r::restart_policy_t::fail_only).spawn();
    }

    auto timeout = shutdown_timeout * 9 / 10;
    create_actor<acceptor_actor_t>().cluster(cluster).timeout(timeout).escalate_failure().finish();
    create_actor<peer_supervisor_t>()
        .cluster(cluster)
        .ssl_pair(&ssl_pair)
        .device_name(app_config.device_name)
        .strand(strand)
        .timeout(timeout)
        .bep_config(app_config.bep_config)
        .relay_config(app_config.relay_config)
        .escalate_failure()
        .finish();
    auto dcfg = app_config.dialer_config;
    if (dcfg.enabled) {
        create_actor<dialer_actor_t>().timeout(timeout).dialer_config(dcfg).cluster(cluster).finish();
    }
}

void net_supervisor_t::on_start() noexcept {
    LOG_TRACE(log, "on_start");
    parent_t::on_start();
    auto timeout = shutdown_timeout * 9 / 10;
    auto io_timeout = shutdown_timeout * 8 / 10;

    create_actor<resolver_actor_t>().timeout(timeout).resolve_timeout(io_timeout).finish();
    create_actor<http_actor_t>()
        .timeout(timeout)
        .request_timeout(io_timeout)
        .resolve_timeout(io_timeout)
        .registry_name(names::http10)
        .keep_alive(false)
        .escalate_failure()
        .finish();
}


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2024 Ivan Baidakou

#pragma once

#include "model/messages.h"
#include "model/cluster.h"
#include "model/misc/sequencer.h"
#include "model/diff/apply_controller.h"
#include "model/diff/cluster_visitor.h"
#include "utils/log.h"
#include "messages.h"
#include <boost/asio.hpp>
#include <rotor/asio.hpp>
#include <boost/outcome.hpp>

namespace syncspirit {
namespace net {

namespace outcome = boost::outcome_v2;

struct net_supervisor_config_t : ra::supervisor_config_asio_t {
    config::main_t app_config;
    size_t cluster_copies = 0;
    model::sequencer_ptr_t sequencer;
};

template <typename Supervisor>
struct net_supervisor_config_builder_t : ra::supervisor_config_asio_builder_t<Supervisor> {
    using builder_t = typename Supervisor::template config_builder_t<Supervisor>;
    using parent_t = ra::supervisor_config_asio_builder_t<Supervisor>;
    using parent_t::parent_t;

    builder_t &&app_config(const config::main_t &value) && noexcept {
        parent_t::config.app_config = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&cluster_copies(size_t value) && noexcept {
        parent_t::config.cluster_copies = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&sequencer(model::sequencer_ptr_t value) && noexcept {
        parent_t::config.sequencer = std::move(value);
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API net_supervisor_t : public ra::supervisor_asio_t,
                                         private model::diff::cluster_visitor_t,
                                         private model::diff::apply_controller_t {
    using parent_t = ra::supervisor_asio_t;
    using config_t = net_supervisor_config_t;

    template <typename Actor> using config_builder_t = net_supervisor_config_builder_t<Actor>;

    explicit net_supervisor_t(config_t &config);
    void configure(r::plugin::plugin_base_t &plugin) noexcept override;
    void on_child_init(actor_base_t *actor, const r::extended_error_ptr_t &ec) noexcept override;
    void on_child_shutdown(actor_base_t *actor) noexcept override;
    void on_start() noexcept override;
    void shutdown_finish() noexcept override;

  private:
    void on_load_cluster(message::load_cluster_response_t &message) noexcept;
    void on_model_update(model::message::model_update_t &message) noexcept;
    void on_model_request(model::message::model_request_t &message) noexcept;

    void dial_peer(const model::device_id_t &peer_device_id, const utils::uri_container_t &uris) noexcept;
    void launch_early() noexcept;
    void launch_cluster() noexcept;
    void launch_net() noexcept;
    void load_db() noexcept;
    void seed_model() noexcept;

    outcome::result<void> save_config(const config::main_t &new_cfg) noexcept;
    outcome::result<void> operator()(const model::diff::load::load_cluster_t &, void *custom) noexcept override;

    model::sequencer_ptr_t sequencer;
    utils::logger_t log;
    config::main_t app_config;
    size_t cluster_copies;
    model::diff::cluster_diff_ptr_t load_diff;
    model::device_id_t global_device;
    r::address_ptr_t db_addr;
    model::cluster_ptr_t cluster;
    utils::key_pair_t ssl_pair;
};

} // namespace net
} // namespace syncspirit


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "peer_actor.h"
#include "names.h"
#include "constants.h"
#include "utils/tls.h"
#include "utils/error_code.h"
#include "utils/format.hpp"
#include "utils/time.h"
#include "transport/stream.h"
#include "proto/bep_support.h"
#include "model/messages.h"
#include "model/diff/contact/peer_state.h"
#include "model/diff/contact/ignored_connected.h"
#include "model/diff/contact/unknown_connected.h"
#include "model/diff/modify/add_pending_device.h"
#include "model/diff/peer/rx_tx.h"

using namespace syncspirit::net;
using namespace syncspirit;

namespace {
namespace resource {
r::plugin::resource_id_t io_read = 0;
r::plugin::resource_id_t io_write = 1;
r::plugin::resource_id_t io_timer = 2;
r::plugin::resource_id_t tx_timer = 3;
r::plugin::resource_id_t rx_timer = 4;
r::plugin::resource_id_t finalization = 5;
} // namespace resource
} // namespace

peer_actor_t::peer_actor_t(config_t &config)
    : r::actor_base_t{config}, cluster{config.cluster}, device_name{config.device_name}, bep_config{config.bep_config},
      coordinator{config.coordinator}, peer_device_id{config.peer_device_id}, transport(std::move(config.transport)),
      peer_endpoint{config.peer_endpoint}, peer_proto(std::move(config.peer_proto)), rx_bytes{0}, tx_bytes{0} {
    rx_buff.resize(config.bep_config.rx_buff_size);
}

void peer_actor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    r::actor_base_t::configure(plugin);

    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        auto id = fmt::format("{}/{}/{}", peer_device_id.get_short(), peer_proto, peer_endpoint);
        p.set_identity(id, false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        p.subscribe_actor(&peer_actor_t::on_start_reading);
        p.subscribe_actor(&peer_actor_t::on_termination);
        p.subscribe_actor(&peer_actor_t::on_block_request);
        p.subscribe_actor(&peer_actor_t::on_transfer);
    });
}

void peer_actor_t::on_start() noexcept {
    r::actor_base_t::on_start();
    LOG_TRACE(log, "on_start");

    auto buff = proto::make_hello_message(device_name);
    push_write(std::move(buff), true, false);

    read_more();
    read_action = &peer_actor_t::read_hello;
    reset_rx_timer();
}

void peer_actor_t::on_io_error(const sys::error_code &ec, rotor::plugin::resource_id_t resource) noexcept {
    LOG_TRACE(log, "on_io_error: {}", ec.message());
    resources->release(resource);
    if (ec != asio::error::operation_aborted) {
        LOG_WARN(log, "on_io_error: {}", ec.message());
    }
    cancel_timer();
    cancel_io();
    if (resources->has(resource::finalization)) {
        resources->release(resource::finalization);
    }
    io_error = true;
    if (state < r::state_t::SHUTTING_DOWN) {
        do_shutdown(make_error(ec));
    }
}

void peer_actor_t::process_tx_queue() noexcept {
    if (finished) {
        return;
    }
    assert(!tx_item);
    if (!tx_queue.empty() && !finished) {
        auto &item = tx_queue.front();
        tx_item = std::move(item);
        tx_queue.pop_front();
        transport::io_fn_t on_write = [&](auto arg) { this->on_write(arg); };
        transport::error_fn_t on_error = [&](auto arg) { this->on_io_error(arg, resource::io_write); };
        auto &tx_buff = tx_item->buff;

        if (tx_buff.size()) {
            resources->acquire(resource::io_write);
            transport->async_send(asio::buffer(tx_buff.data(), tx_buff.size()), on_write, on_error);
        } else {
            LOG_TRACE(log, "peer_actor_t::process_tx_queue, device_id = {}, final empty message, shutting down ",
                      peer_device_id);
            assert(tx_item->final);
            auto ec = r::make_error_code(r::shutdown_code_t::normal);
            do_shutdown(make_error(ec));
            finished = true;
        }
    }
}

void peer_actor_t::push_write(utils::bytes_t buff, bool signal, bool final) noexcept {
    if (io_error) {
        return;
    }
    if (signal && controller) {
        send<payload::transfer_push_t>(controller, buff.size());
    }
    tx_item_t item = new confidential::payload::tx_item_t{std::move(buff), final};
    tx_queue.emplace_back(std::move(item));
    if (!tx_item) {
        process_tx_queue();
    }
    if (final) {
        resources->acquire(resource::finalization);
    }
}

void peer_actor_t::read_more() noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }
    if (rx_idx >= rx_buff.size()) {
        LOG_WARN(log, "read_more, rx buffer limit reached, {}", rx_buff.size());
        auto ec = utils::make_error_code(utils::error_code_t::rx_limit_reached);
        return do_shutdown(make_error(ec));
    }

    transport::io_fn_t on_read = [&](auto arg) { this->on_read(arg); };
    transport::error_fn_t on_error = [&](auto arg) { this->on_io_error(arg, resource::io_read); };
    resources->acquire(resource::io_read);
    auto buff = asio::buffer(rx_buff.data() + rx_idx, rx_buff.size() - rx_idx);
    LOG_TRACE(log, "read_more");
    transport->async_recv(buff, on_read, on_error);
}

void peer_actor_t::on_write(std::size_t sz) noexcept {
    resources->release(resource::io_write);
    LOG_TRACE(log, "on_write, {} bytes", sz);
    if (controller) {
        send<payload::transfer_pop_t>(controller, (uint32_t)sz);
    }
    tx_bytes += sz;
    emit_io_stats();

    assert(tx_item);
    if (tx_item->final) {
        LOG_TRACE(log, "process_tx_queue, final message has been sent, shutting down");
        if (resources->has(resource::finalization)) {
            resources->release(resource::finalization);
        }
        cancel_io();
    } else {
        tx_item.reset();
        process_tx_queue();
    }
}

void peer_actor_t::on_read(std::size_t bytes) noexcept {
    assert(read_action);
    resources->release(resource::io_read);
    rx_idx += bytes;

    LOG_TRACE(log, "on_read, {} bytes, total = {}", bytes, rx_idx);
    rx_bytes += bytes;

    auto buff = utils::bytes_view_t((unsigned char *)rx_buff.data(), rx_idx);
    auto result = proto::parse_bep(buff);
    if (result.has_error()) {
        auto &ec = result.error();
        LOG_WARN(log, "on_read, error parsing message: {}", ec.message());
        return do_shutdown(make_error(ec));
    }
    auto &value = result.value();
    if (!value.consumed) {
        // log->trace("on_read :: incomplete message");
        return read_more();
    }

    cancel_timer();
    assert(value.consumed <= rx_idx);
    rx_idx -= value.consumed;
    if (rx_idx) {
        auto ptr = rx_buff.data();
        std::memcpy(ptr, ptr + value.consumed, rx_idx);
    }
    (this->*read_action)(std::move(value.message));
    emit_io_stats();
    // LOG_TRACE(log, "on_read, rx_idx = {} ", rx_idx);
}

void peer_actor_t::on_timer(r::request_id_t, bool cancelled) noexcept {
    resources->release(resource::io_timer);
    LOG_TRACE(log, "on_timer_trigger, cancelled = {}", cancelled);
    if (!cancelled) {
        cancel_io();
        auto ec = r::make_error_code(r::shutdown_code_t::normal);
        do_shutdown(make_error(ec));
    }
}

void peer_actor_t::shutdown_start() noexcept {
    LOG_TRACE(log, "shutdown_start");
    if (resources->has(resource::io_timer)) {
        cancel_timer();
    }
    if (tx_timer_request) {
        r::actor_base_t::cancel_timer(*tx_timer_request);
    }
    if (rx_timer_request) {
        r::actor_base_t::cancel_timer(*rx_timer_request);
    }
    if (controller) {
        send<payload::termination_t>(controller, shutdown_reason);
    }

    proto::Close close;
    proto::set_reason(close, shutdown_reason->message());
    auto buff = proto::serialize(close);
    tx_queue.clear();
    push_write(std::move(buff), true, true);
    LOG_TRACE(log, "going to send close message");

    r::actor_base_t::shutdown_start();
}

void peer_actor_t::shutdown_finish() noexcept {
    LOG_TRACE(log, "shutdown_finish");
    for (auto &it : block_requests) {
        auto ec = r::make_error_code(r::error_code_t::cancelled);
        reply_with_error(*it, make_error(ec));
    }
    block_requests.clear();
    if (controller) {
        send<payload::termination_t>(controller, shutdown_reason);
    }
    r::actor_base_t::shutdown_finish();
    auto sha256 = peer_device_id.get_sha256();
    auto device = cluster->get_devices().by_sha256(sha256);
    if (device && device->get_state() != model::device_state_t::offline) {
        auto state = model::device_state_t::offline;
        auto connection_id = fmt::format("{}://{}", peer_proto, peer_endpoint);
        auto diff = model::diff::contact::peer_state_t::create(*cluster, sha256, address, state, connection_id);
        if (diff) {
            send<model::payload::model_update_t>(coordinator, std::move(diff));
        }
    }
}

void peer_actor_t::cancel_timer() noexcept {
    if (resources->has(resource::io_timer)) {
        r::actor_base_t::cancel_timer(*timer_request);
        timer_request.reset();
    }
}

void peer_actor_t::cancel_io() noexcept {
    if (resources->has(resource::io_read)) {
        LOG_TRACE(log, "cancelling I/O (read)");
        transport->cancel();
        return;
    }
    if (resources->has(resource::io_write)) {
        LOG_TRACE(log, "cancelling I/O (write)");
        transport->cancel();
        return;
    }
}

void peer_actor_t::on_start_reading(message::start_reading_t &message) noexcept {
    bool start = message.payload.start;
    LOG_TRACE(log, "on_start_reading, start = ", start);
    controller = message.payload.controller;
    if (start) {
        read_action = &peer_actor_t::read_controlled;
        read_more();
    }
}

void peer_actor_t::on_termination(message::termination_signal_t &message) noexcept {
    if (!shutdown_reason) {
        auto &ee = message.payload.ee;
        auto reason = ee->message();
        LOG_DEBUG(log, "on_termination: {}", reason);
        do_shutdown(ee);
    }
}

void peer_actor_t::on_block_request(message::block_request_t &message) noexcept {
    auto req_id = (std::int32_t)message.payload.id;
    auto &p = message.payload.request_payload;
    proto::Request req;
    proto::set_id(req, req_id);
    proto::set_folder(req, p.folder_id);
    proto::set_name(req, p.file_name);
    proto::set_offset(req, p.block_offset);
    proto::set_size(req, p.block_size);
    proto::set_hash(req, p.block_hash);

    auto buff = proto::serialize(req);
    push_write(std::move(buff), true, false);
    block_requests.emplace_back(&message);
}

void peer_actor_t::on_transfer(message::transfer_data_t &message) noexcept {
    LOG_TRACE(log, "on_transfer");

    push_write(std::move(message.payload.data), false, false);
}

void peer_actor_t::read_hello(proto::message::message_t &&msg) noexcept {
    LOG_TRACE(log, "read_hello");
    std::visit(
        [&](auto &&msg) {
            using T = std::decay_t<decltype(msg)>;
            if constexpr (std::is_same_v<T, proto::Hello>) {
                return handle_hello(std::move(msg));
            } else {
                LOG_WARN(log, "read_hello, unexpected_message");
                auto ec = utils::make_error_code(utils::bep_error_code_t::unexpected_message);
                return do_shutdown(make_error(ec));
            }
        },
        msg);
}

void peer_actor_t::read_controlled(proto::message::message_t &&msg) noexcept {
    using MT = proto::MessageType;
    LOG_TRACE(log, "read_controlled");
    auto type = MT::UNKNOWN;
    std::visit(
        [&](auto &&msg) {
            using T = std::decay_t<decltype(msg)>;
            type = proto::message::get_bep_type<T>();
            if constexpr (std::is_same_v<T, proto::Hello>) {
                LOG_WARN(log, "{}, hello, unexpected_message");
                auto ec = utils::make_error_code(utils::bep_error_code_t::unexpected_message);
                return do_shutdown(make_error(ec));
            } else if constexpr (std::is_same_v<T, proto::Ping>) {
                handle_ping(std::move(msg));
            } else if constexpr (std::is_same_v<T, proto::Close>) {
                handle_close(std::move(msg));
            } else if constexpr (std::is_same_v<T, proto::Response>) {
                handle_response(std::move(msg));
            } else {
                auto fwd = payload::forwarded_message_t{std::move(msg)};
                send<payload::forwarded_message_t>(controller, std::move(fwd));
                reset_rx_timer();
            }
        },
        msg);
    LOG_DEBUG(log, "read_controlled, type = {}", (int)type);
    bool continue_reading = !((type == MT::HELLO) || (type == MT::UNKNOWN) || (type == MT::CLOSE));
    if (continue_reading) {
        read_action = &peer_actor_t::read_controlled;
        read_more();
    }
}

void peer_actor_t::handle_hello(proto::Hello &&msg) noexcept {
    using namespace model::diff;
    auto device_name = proto::get_device_name(msg);
    auto client_name = proto::get_client_name(msg);
    auto client_version = proto::get_client_version(msg);
    auto sha_s256 = peer_device_id.get_sha256();
    LOG_DEBUG(log, "read_hello, from {} ({} {})", device_name, client_name, client_version);
    auto diff = cluster_diff_ptr_t();
    auto db = db::SomeDevice();
    auto fill_db_and_shutdown = [&]() {
        auto address = fmt::format("{}://{}", peer_proto, peer_endpoint);
        db::set_name(db, device_name);
        db::set_client_name(db, client_name);
        db::set_client_version(db, client_version);
        db::set_address(db, std::move(address));
        db::set_last_seen(db, utils::as_seconds(pt::microsec_clock::local_time()));
        LOG_INFO(log, "device {} is unknown/ignored, shutting down", device_name);
        do_shutdown();
    };

    if (auto known_peer = cluster->get_devices().by_sha256(sha_s256); known_peer) {
        if (known_peer->get_state() == model::device_state_t::online) {
            auto ec = utils::make_error_code(utils::error_code_t::already_connected);
            return do_shutdown(make_error(ec));
        }
        auto state = model::device_state_t::online;
        auto connection_id = fmt::format("{}://{}", peer_proto, peer_endpoint);
        diff = contact::peer_state_t::create(*cluster, sha_s256, get_address(), state, std::move(connection_id),
                                             cert_name, peer_endpoint, client_name, client_version);
    } else if (auto peer = cluster->get_ignored_devices().by_sha256(sha_s256)) {
        fill_db_and_shutdown();
        diff = new model::diff::contact::ignored_connected_t(*cluster, peer_device_id, std::move(db));
    } else {
        fill_db_and_shutdown();
        if (auto peer = cluster->get_pending_devices().by_sha256(sha_s256); !peer) {
            auto cluster_diff = cluster_diff_ptr_t{};
            cluster_diff = new model::diff::modify::add_pending_device_t(peer_device_id, db);
            send<model::payload::model_update_t>(coordinator, std::move(cluster_diff));
        }
        // send add unknown device diff
        diff = new model::diff::contact::unknown_connected_t(*cluster, peer_device_id, std::move(db));
    }

    if (diff) {
        send<model::payload::model_update_t>(coordinator, std::move(diff));
    }
}

void peer_actor_t::handle_ping(proto::Ping &&) noexcept { log->trace("handle_ping"); }

void peer_actor_t::handle_close(proto::Close &&message) noexcept {
    auto reason = proto::get_reason(message);
    const char *str = reason.data();
    LOG_TRACE(log, "handle_close, reason = {}", reason);
    if (reason.size() == 0) {
        str = "no reason specified";
    }
    auto ee = r::make_error(str, r::shutdown_code_t::normal);
    do_shutdown(ee);
}

void peer_actor_t::handle_response(proto::Response &&message) noexcept {
    auto id = proto::get_id(message);
    LOG_TRACE(log, "handle_response, message id = {}", id);
    auto predicate = [id = id](const block_request_ptr_t &it) { return ((std::int32_t)it->payload.id) == id; };
    auto it = std::find_if(block_requests.begin(), block_requests.end(), predicate);
    if (it == block_requests.end()) {
        if (!shutdown_reason) {
            LOG_WARN(log, "response for unexpected request id {}", id);
            auto ec = utils::make_error_code(utils::bep_error_code_t::response_mismatch);
            do_shutdown(make_error(ec));
        }
    }

    auto error = proto::get_code(message);
    auto &block_request = *it;
    if (!shutdown_reason) {
        if (error != proto::ErrorCode::NO_BEP_ERROR) {
            auto ec = utils::make_error_code((utils::request_error_code_t)error);
            LOG_WARN(log, "block request error: {}", ec.message());
            reply_with_error(*block_request, make_error(ec));
        } else {
            auto data = proto::extract_data(message);
            auto request_sz = block_request->payload.request_payload.block_size;
            if (data.size() != request_sz) {
                LOG_WARN(log, "got {} bytes, but requested {}", data.size(), request_sz);
                auto ec = utils::make_error_code(utils::bep_error_code_t::response_missize);
                return do_shutdown(make_error(ec));
            }
            auto bytes = utils::bytes_t(std::move(data));
            reply_to(*block_request, std::move(bytes));
        }
    }
    block_requests.erase(it);
}

void peer_actor_t::reset_tx_timer() noexcept {
    if (state == r::state_t::OPERATIONAL) {
        if (tx_timer_request) {
            r::actor_base_t::cancel_timer(*tx_timer_request);
        }
        auto timeout = pt::milliseconds(bep_config.tx_timeout);
        tx_timer_request = start_timer(timeout, *this, &peer_actor_t::on_tx_timeout);
        resources->acquire(resource::tx_timer);
    }
}

void peer_actor_t::on_tx_timeout(r::request_id_t, bool cancelled) noexcept {
    resources->release(resource::tx_timer);
    tx_timer_request.reset();
    if (!cancelled) {
        proto::Ping ping;
        auto buff = proto::serialize(ping);
        push_write(std::move(buff), true, false);
        reset_tx_timer();
    }
}

void peer_actor_t::reset_rx_timer() noexcept {
    if (state == r::state_t::OPERATIONAL) {
        if (rx_timer_request) {
            r::actor_base_t::cancel_timer(*rx_timer_request);
        }
        auto timeout = pt::milliseconds(bep_config.rx_timeout);
        rx_timer_request = start_timer(timeout, *this, &peer_actor_t::on_rx_timeout);
        resources->acquire(resource::rx_timer);
    }
}

void peer_actor_t::on_rx_timeout(r::request_id_t, bool cancelled) noexcept {
    resources->release(resource::rx_timer);
    rx_timer_request.reset();
    if (!cancelled) {
        auto ec = utils::make_error_code(utils::error_code_t::rx_timeout);
        auto reason = make_error(ec);
        do_shutdown(reason);
    }
}

void peer_actor_t::emit_io_stats() noexcept {
    using namespace std::chrono;
    auto now = clock_t::now();
    if (now > last_stats + milliseconds{bep_config.stats_interval}) {
        auto diff = model::diff::cluster_diff_ptr_t();
        last_stats = now;
        diff.reset(new model::diff::peer::rx_tx_t(peer_device_id.get_sha256(), rx_bytes, tx_bytes));
        rx_bytes = tx_bytes = 0;
        send<model::payload::model_update_t>(coordinator, std::move(diff));
    }
}


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#pragma once

#include "config/bep.h"
#include "proto/bep_support.h"
#include "utils/log.h"
#include "model/cluster.h"
#include "messages.h"
#include <boost/asio.hpp>
#include <rotor/asio/supervisor_asio.h>
#include <optional>
#include <list>
#include <chrono>

namespace syncspirit {

namespace net {

struct peer_actor_config_t : public r::actor_config_t {
    std::string_view device_name;
    model::device_id_t peer_device_id;
    config::bep_config_t bep_config;
    transport::stream_sp_t transport;
    r::address_ptr_t coordinator;
    model::cluster_ptr_t cluster;
    tcp::endpoint peer_endpoint;
    std::string peer_proto;
};

template <typename Actor> struct peer_actor_config_builder_t : r::actor_config_builder_t<Actor> {
    using builder_t = typename Actor::template config_builder_t<Actor>;
    using parent_t = r::actor_config_builder_t<Actor>;
    using parent_t::parent_t;

    builder_t &&peer_device_id(const model::device_id_t &value) && noexcept {
        parent_t::config.peer_device_id = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&device_name(std::string_view value) && noexcept {
        parent_t::config.device_name = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&bep_config(const config::bep_config_t &value) && noexcept {
        parent_t::config.bep_config = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&coordinator(const r::address_ptr_t &value) && noexcept {
        parent_t::config.coordinator = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&cluster(const model::cluster_ptr_t &value) && noexcept {
        parent_t::config.cluster = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&transport(transport::stream_sp_t value) && noexcept {
        parent_t::config.transport = std::move(value);
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&peer_endpoint(const tcp::endpoint &value) && noexcept {
        parent_t::config.peer_endpoint = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&peer_proto(std::string value) && noexcept {
        parent_t::config.peer_proto = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API peer_actor_t : public r::actor_base_t {
    using config_t = peer_actor_config_t;
    template <typename Actor> using config_builder_t = peer_actor_config_builder_t<Actor>;

    peer_actor_t(config_t &config);
    void configure(r::plugin::plugin_base_t &plugin) noexcept override;
    void on_start() noexcept override;
    void shutdown_start() noexcept override;
    void shutdown_finish() noexcept override;

  private:
    struct confidential {
        struct payload {
            struct tx_item_t : boost::intrusive_ref_counter<tx_item_t, boost::thread_unsafe_counter> {
                utils::bytes_t buff;
                bool final = false;

                tx_item_t(utils::bytes_t buff_, bool final_) noexcept : buff{std::move(buff_)}, final{final_} {}
                tx_item_t(tx_item_t &&other) = default;
            };
        };

        struct message {
            using tx_item_t = r::message_t<payload::tx_item_t>;
        };
    };

    using tx_item_t = model::intrusive_ptr_t<confidential::payload::tx_item_t>;
    using tx_message_t = confidential::message::tx_item_t;
    using tx_queue_t = std::list<tx_item_t>;
    using read_action_t = void (peer_actor_t::*)(proto::message::message_t &&msg);
    using block_request_ptr_t = r::intrusive_ptr_t<message::block_request_t>;
    using block_requests_t = std::list<block_request_ptr_t>;
    using clock_t = std::chrono::steady_clock;

    void on_start_reading(message::start_reading_t &) noexcept;
    void on_termination(message::termination_signal_t &) noexcept;
    void on_block_request(message::block_request_t &) noexcept;
    void on_transfer(message::transfer_data_t &message) noexcept;

    void on_io_error(const sys::error_code &ec, r::plugin::resource_id_t resource) noexcept;
    void on_write(std::size_t bytes) noexcept;
    void on_read(std::size_t bytes) noexcept;
    void on_timer(r::request_id_t, bool cancelled) noexcept;
    void read_more() noexcept;
    void push_write(utils::bytes_t buff, bool signal, bool final) noexcept;
    void process_tx_queue() noexcept;
    void cancel_timer() noexcept;
    void cancel_io() noexcept;
    void on_tx_timeout(r::request_id_t, bool cancelled) noexcept;
    void on_rx_timeout(r::request_id_t, bool cancelled) noexcept;

    void reset_tx_timer() noexcept;
    void reset_rx_timer() noexcept;
    void read_hello(proto::message::message_t &&msg) noexcept;
    void read_controlled(proto::message::message_t &&msg) noexcept;

    void handle_hello(proto::Hello &&) noexcept;
    void handle_ping(proto::Ping &&) noexcept;
    void handle_close(proto::Close &&) noexcept;
    void handle_response(proto::Response &&) noexcept;

    void emit_io_stats() noexcept;

    model::cluster_ptr_t cluster;
    utils::logger_t log;
    std::string_view device_name;
    config::bep_config_t bep_config;
    r::address_ptr_t coordinator;
    model::device_id_t peer_device_id;
    transport::stream_sp_t transport;
    std::optional<r::request_id_t> timer_request;
    std::optional<r::request_id_t> tx_timer_request;
    std::optional<r::request_id_t> rx_timer_request;
    tx_queue_t tx_queue;
    tx_item_t tx_item;
    fmt::memory_buffer rx_buff;
    std::size_t rx_idx = 0;
    std::string cert_name;
    tcp::endpoint peer_endpoint;
    std::string peer_proto;
    read_action_t read_action;
    r::address_ptr_t controller;
    block_requests_t block_requests;
    std::size_t rx_bytes;
    std::size_t tx_bytes;
    clock_t::time_point last_stats;
    bool finished = false;
    bool io_error = false;
};

} // namespace net
} // namespace syncspirit


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "peer_supervisor.h"
#include "peer_actor.h"
#include "initiator_actor.h"
#include "names.h"
#include "../constants.h"
#include "utils/format.hpp"
#include "model/diff/contact/peer_state.h"
#include "model/diff/contact/connect_request.h"
#include "model/diff/contact/relay_connect_request.h"
#include "model/diff/contact/dial_request.h"
#include "model/misc/error_code.h"

using namespace syncspirit::net;

template <class> inline constexpr bool always_false_v = false;

peer_supervisor_t::peer_supervisor_t(peer_supervisor_config_t &cfg)
    : parent_t{cfg}, cluster{cfg.cluster}, device_name{cfg.device_name}, ssl_pair{*cfg.ssl_pair},
      bep_config(cfg.bep_config), relay_config{cfg.relay_config} {}

void peer_supervisor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    r::actor_base_t::configure(plugin);
    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        p.set_identity(names::peer_supervisor, false);
        log = utils::get_logger(identity);
        addr_unknown = p.create_address();
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.register_name(names::peer_supervisor, get_address());
        p.discover_name(names::coordinator, coordinator, true).link(false).callback([&](auto phase, auto &ee) {
            if (!ee && phase == r::plugin::registry_plugin_t::phase_t::linking) {
                auto p = get_plugin(r::plugin::starter_plugin_t::class_identity);
                auto plugin = static_cast<r::plugin::starter_plugin_t *>(p);
                plugin->subscribe_actor(&peer_supervisor_t::on_model_update, coordinator);
                plugin->subscribe_actor(&peer_supervisor_t::on_peer_ready);
                plugin->subscribe_actor(&peer_supervisor_t::on_connect);
                plugin->subscribe_actor(&peer_supervisor_t::on_connected, addr_unknown);
            }
        });
    });
}

void peer_supervisor_t::on_child_shutdown(actor_base_t *actor) noexcept {
    using namespace model::diff;
    auto &reason = actor->get_shutdown_reason();
    LOG_TRACE(log, "on_child_shutdown, {} due to {} ", actor->get_identity(), reason->message());
    parent_t::on_child_shutdown(actor);
}

void peer_supervisor_t::on_start() noexcept {
    LOG_TRACE(log, "on_start");
    parent_t::on_start();
}

void peer_supervisor_t::on_model_update(model::message::model_update_t &msg) noexcept {
    LOG_TRACE(log, "on_model_update");
    auto &diff = *msg.payload.diff;
    auto r = diff.visit(*this, nullptr);
    if (!r) {
        auto ee = make_error(r.assume_error());
        do_shutdown(ee);
    }
}

void peer_supervisor_t::on_peer_ready(message::peer_connected_t &msg) noexcept {
    LOG_TRACE(log, "on_peer_ready");
    auto timeout = r::pt::milliseconds{bep_config.connect_timeout};
    auto &p = msg.payload;
    auto &d = p.peer_device_id;
    auto peer = cluster->get_devices().by_sha256(d.get_sha256());
    if (peer && peer->get_state() == model::device_state_t::online) {
        LOG_DEBUG(log, "peer '{}' is already online, ignoring request", d.get_short());
        return;
    }
    create_actor<peer_actor_t>()
        .transport(std::move(p.transport))
        .peer_device_id(d)
        .device_name(device_name)
        .bep_config(bep_config)
        .coordinator(coordinator)
        .peer_endpoint(p.remote_endpoint)
        .peer_proto(p.proto)
        .timeout(timeout)
        .cluster(cluster)
        .finish();
}

void peer_supervisor_t::on_connected(message::peer_connected_t &msg) noexcept {
    LOG_TRACE(log, "on_connected");
    auto &p = msg.payload;
    auto req = static_cast<message::connect_request_t *>(p.custom.get());
    reply_to(*req, std::move(p.transport), std::move(p.remote_endpoint));
}

void peer_supervisor_t::on_connect(message::connect_request_t &msg) noexcept {
    auto &p = msg.payload.request_payload;
    auto connect_timeout = r::pt::milliseconds{bep_config.connect_timeout};
    create_actor<initiator_actor_t>()
        .ssl_pair(&ssl_pair)
        .peer_device_id(p.device_id)
        .uris({p.uri})
        .custom(&msg)
        .router(*locality_leader)
        .sink(addr_unknown)
        .alpn(p.alpn)
        .init_timeout(connect_timeout * 2)
        .shutdown_timeout(connect_timeout)
        .finish();
}

auto peer_supervisor_t::operator()(const model::diff::contact::connect_request_t &diff, void *custom) noexcept
    -> outcome::result<void> {

    auto lock = std::unique_lock(diff.mutex);
    auto sock = std::move(*diff.sock);
    diff.sock.reset();

    auto timeout = r::pt::milliseconds{bep_config.connect_timeout};
    create_actor<initiator_actor_t>()
        .cluster(cluster)
        .router(*locality_leader)
        .sink(address)
        .ssl_pair(&ssl_pair)
        .sock(std::move(sock))
        .alpn(constants::protocol_name)
        .timeout(timeout)
        .finish();
    return diff.visit_next(*this, custom);
}

auto peer_supervisor_t::operator()(const model::diff::contact::relay_connect_request_t &diff, void *custom) noexcept
    -> outcome::result<void> {

    auto peer = cluster->get_devices().by_sha256(diff.peer.get_sha256());
    if (peer->get_state() == model::device_state_t::offline) {
        LOG_DEBUG(log, "initiating relay connection with {}", peer->device_id());
        auto timeout = r::pt::milliseconds{bep_config.connect_timeout};
        auto uri_str = fmt::format("tcp://{}", diff.relay);
        auto uri = utils::parse(uri_str);
        create_actor<initiator_actor_t>()
            .cluster(cluster)
            .router(*locality_leader)
            .sink(address)
            .ssl_pair(&ssl_pair)
            .peer_device_id(diff.peer)
            .uris(utils::uri_container_t{std::move(uri)})
            .relay_session(diff.session_key)
            .timeout(timeout)
            .finish();
    } else {
        LOG_DEBUG(log, "peer '{}' is not offline, dropping relay connection request", peer->device_id());
    }
    return diff.visit_next(*this, custom);
}

auto peer_supervisor_t::operator()(const model::diff::contact::dial_request_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto &devices = cluster->get_devices();
    auto peer = devices.by_sha256(diff.peer_id);
    if (peer && peer->get_state() != model::device_state_t::online) {
        auto &uris = peer->get_uris();
        LOG_DEBUG(log, "initiating connection with {} ({} addresses)", peer->device_id(), uris.size());
        assert(uris.size());
        auto connect_timeout = r::pt::milliseconds{bep_config.connect_timeout};
        create_actor<initiator_actor_t>()
            .router(*locality_leader)
            .sink(address)
            .ssl_pair(&ssl_pair)
            .peer_device_id(peer->device_id())
            .uris(uris)
            .relay_enabled(relay_config.enabled)
            .cluster(cluster)
            .init_timeout(connect_timeout * (uris.size() + 1))
            .shutdown_timeout(connect_timeout)
            .finish();
    }
    return diff.visit_next(*this, custom);
}


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2024 Ivan Baidakou

#pragma once

#include "config/bep.h"
#include "config/relay.h"
#include "messages.h"
#include "model/messages.h"
#include "model/diff/cluster_visitor.h"
#include "utils/log.h"
#include <boost/asio.hpp>
#include <rotor/asio.hpp>

namespace syncspirit {
namespace net {

namespace payload {
struct peer_connected_t;
}

namespace message {
using peer_connected_t = r::message_t<payload::peer_connected_t>;
}

namespace outcome = boost::outcome_v2;

struct peer_supervisor_config_t : ra::supervisor_config_asio_t {
    std::string_view device_name;
    const utils::key_pair_t *ssl_pair;
    config::bep_config_t bep_config;
    config::relay_config_t relay_config;
    model::cluster_ptr_t cluster;
};

template <typename Supervisor>
struct peer_supervisor_config_builder_t : ra::supervisor_config_asio_builder_t<Supervisor> {
    using builder_t = typename Supervisor::template config_builder_t<Supervisor>;
    using parent_t = ra::supervisor_config_asio_builder_t<Supervisor>;
    using parent_t::parent_t;

    builder_t &&device_name(std::string_view value) && noexcept {
        parent_t::config.device_name = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&ssl_pair(const utils::key_pair_t *value) && noexcept {
        parent_t::config.ssl_pair = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&bep_config(const config::bep_config_t &value) && noexcept {
        parent_t::config.bep_config = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&relay_config(const config::relay_config_t &value) && noexcept {
        parent_t::config.relay_config = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&cluster(const model::cluster_ptr_t &value) && noexcept {
        parent_t::config.cluster = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API peer_supervisor_t : public ra::supervisor_asio_t, private model::diff::cluster_visitor_t {
    using parent_t = ra::supervisor_asio_t;
    using config_t = peer_supervisor_config_t;
    template <typename Actor> using config_builder_t = peer_supervisor_config_builder_t<Actor>;

    explicit peer_supervisor_t(peer_supervisor_config_t &config);
    void configure(r::plugin::plugin_base_t &plugin) noexcept override;
    void on_child_shutdown(actor_base_t *actor) noexcept override;
    void on_start() noexcept override;

  private:
    void on_connect(message::connect_request_t &) noexcept;
    void on_model_update(model::message::model_update_t &) noexcept;
    void on_peer_ready(message::peer_connected_t &) noexcept;
    void on_connected(message::peer_connected_t &) noexcept;

    outcome::result<void> operator()(const model::diff::contact::dial_request_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::contact::connect_request_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::contact::relay_connect_request_t &, void *) noexcept override;

    model::cluster_ptr_t cluster;
    utils::logger_t log;
    r::address_ptr_t coordinator;
    r::address_ptr_t addr_unknown;
    std::string_view device_name;
    const utils::key_pair_t &ssl_pair;
    config::bep_config_t bep_config;
    config::relay_config_t relay_config;
};

} // namespace net
} // namespace syncspirit


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "relay_actor.h"
#include "names.h"
#include "constants.h"
#include "utils/error_code.h"
#include "utils/beast_support.h"
#include "utils/time.h"
#include "model/messages.h"
#include "model/diff/contact/update_contact.h"
#include "model/diff/contact/relay_connect_request.h"
#include "model/diff/contact/unknown_connected.h"
#include "model/diff/modify/add_pending_device.h"
#include "transport/stream.h"
#include <spdlog/fmt/bin_to_hex.h>
#include "messages.h"
#include <cstdlib>
#include <type_traits>
#include <boost/core/demangle.hpp>

using namespace syncspirit::net;

static const constexpr size_t BUFF_SZ = 1500;
template <class> inline constexpr bool always_false_v = false;

namespace {
namespace resource {
r::plugin::resource_id_t init = 0;
r::plugin::resource_id_t http = 1;
r::plugin::resource_id_t io_read = 2;
r::plugin::resource_id_t io_write = 3;
} // namespace resource
} // namespace

relay_actor_t::relay_actor_t(config_t &config) noexcept
    : r::actor_base_t(config), cluster{std::move(config.cluster)}, config{config.config} {
    http_rx_buff = std::make_shared<payload::http_request_t::rx_buff_t>();
}

void relay_actor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    r::actor_base_t::configure(plugin);
    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        p.set_identity("net.relay", false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.discover_name(names::http11_relay, http_client, true).link(true);
        p.discover_name(names::coordinator, coordinator, false).link(false);
        p.discover_name(names::peer_supervisor, peer_supervisor, true).link(false);
    });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        p.subscribe_actor(&relay_actor_t::on_list);
        p.subscribe_actor(&relay_actor_t::on_connect);
        resources->acquire(resource::init);
        request_relay_list();
    });
}

void relay_actor_t::shutdown_start() noexcept {
    LOG_TRACE(log, "shutdown_start");
    r::actor_base_t::shutdown_start();

    if (resources->has(resource::init)) {
        resources->release(resource::init);
    }
    if (resources->has(resource::http)) {
        send<message::http_cancel_t::payload_t>(http_client, *http_request, get_address());
    }
    if (resources->has(resource::io_read) || resources->has(resource::io_write)) {
        master->cancel();
    }
    if (rx_state) {
        auto self = cluster->get_device();
        utils::uri_container_t uris;
        for (auto &uri : self->get_uris()) {
            if (uri->scheme() != "relay") {
                uris.emplace_back(uri);
            }
        }
        using namespace model::diff;
        auto diff = model::diff::cluster_diff_ptr_t{};
        diff = new contact::update_contact_t(*cluster, self->device_id(), uris);
        send<model::payload::model_update_t>(coordinator, std::move(diff), this);
    }
}

void relay_actor_t::on_start() noexcept {
    LOG_TRACE(log, "on_start");
    r::actor_base_t::on_start();
    connect_to_relay();
}

void relay_actor_t::connect_to_relay() noexcept {
    size_t attempts = 0;
    while (++attempts < 10) {
        relay_index = rand() % relays.size();
        auto relay = relays[relay_index];
        if (!relay) {
            continue;
        }
        auto &l = relay->location;
        auto relay_host = relay->uri->host();
        auto relay_port = std::string_view(relay->uri->port());
        LOG_INFO(log, "chosen relay({}) {}:{}, city: {}, country: {}, continent: {}", relay_index, relay_host,
                 relay_port, l.city, l.country, l.continent);

        auto uri = utils::parse(fmt::format("tcp://{}:{}", relay_host, relay_port));
        request<payload::connect_request_t>(peer_supervisor, relay->device_id, std::move(uri),
                                            constants::relay_protocol_name)
            .send(init_timeout);
        return;
    }
    if (attempts >= 10) {
        auto ec = utils::make_error_code(utils::error_code_t::relay_failure);
        do_shutdown(make_error(ec));
    }
}

void relay_actor_t::request_relay_list() noexcept {
    auto timeout = init_timeout * 9 / 10;
    auto &uri = config.discovery_url;
    LOG_TRACE(log, "going to request relays list, timeout = {} ms, via {}:{}", timeout.total_milliseconds(),
              std::string_view(uri->host()), uri->port_number());

    assert(uri->scheme() == "https");
    http::request<http::empty_body> req;
    req.method(http::verb::get);
    req.version(10);
    req.target(uri->encoded_target());
    req.set(http::field::host, uri->host());
    req.set(http::field::connection, "close");

    utils::bytes_t tx_buff;
    auto r = utils::serialize(req, tx_buff);
    if (!r) {
        auto &ec = r.assume_error();
        LOG_WARN(log, "cannot serialize request: {}'", r.assume_error().message());
        return do_shutdown(make_error(ec));
    }
    resources->acquire(resource::http);
    transport::ssl_junction_t ssl{model::device_id_t{}, nullptr, true, {}};
    http_request = request<payload::http_request_t>(http_client, uri, std::move(tx_buff), http_rx_buff,
                                                    config.rx_buff_size, std::move(ssl), config.debug)
                       .send(timeout);
}

void relay_actor_t::on_list(message::http_response_t &msg) noexcept {
    LOG_TRACE(log, "on_list");
    resources->release(resource::http);

    auto &ee = msg.payload.ee;
    if (ee) {
        LOG_WARN(log, "get public relays failed: {}", ee->message());
        auto inner = utils::make_error_code(utils::error_code_t::cannot_get_public_relays);
        return do_shutdown(make_error(inner, ee));
    }
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    auto &body = msg.payload.res->response.body();
    auto result = proto::relay::parse_endpoint(body);
    if (!result) {
        auto &ec = result.assume_error();
        LOG_WARN(log, "cannot parse relays: {}", ec.message());
        return do_shutdown(make_error(ec));
    }
    auto &list = result.assume_value();
    if (list.empty()) {
        LOG_WARN(log, "empty list of public relays");
        auto ec = utils::make_error_code(utils::error_code_t::cannot_get_public_relays);
        return do_shutdown(make_error(ec));
    }
    relays = std::move(list);
    http_rx_buff.reset();
    resources->release(resource::init);
}

void relay_actor_t::read_master() noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }
    transport::io_fn_t on_read = [&](auto arg) { this->on_read(arg); };
    transport::error_fn_t on_error = [&](auto arg) { this->on_io_error(arg, resource::io_read); };
    resources->acquire(resource::io_read);
    auto buff = asio::buffer(rx_buff.data() + rx_idx, rx_buff.size() - rx_idx);
    LOG_TRACE(log, "read_master, sz = {}", buff.size());
    master->async_recv(buff, on_read, on_error);
}

void relay_actor_t::push_master(utils::bytes_t data) noexcept {
    tx_queue.emplace_back(tx_item_t(new utils::bytes_t(std::move(data))));
    if (!resources->has(resource::io_write)) {
        write_master();
    }
}

void relay_actor_t::write_master() noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }
    auto &item = tx_queue.front();
    transport::io_fn_t on_write = [&](auto sz) {
        LOG_TRACE(log, "write {} bytes", sz);
        resources->release(resource::io_write);
        tx_queue.pop_front();
        if (tx_timer) {
            cancel_timer(*tx_timer);
        }
        if (!tx_queue.empty()) {
            write_master();
        }
    };
    transport::error_fn_t on_error = [&](auto arg) { this->on_io_error(arg, resource::io_write); };
    auto buff = asio::buffer(*item);
    master->async_send(buff, on_write, on_error);
    resources->acquire(resource::io_write);
    respawn_tx_timer();
}

void relay_actor_t::on_connect(message::connect_response_t &res) noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }
    LOG_TRACE(log, "on_connect");
    auto &ee = res.payload.ee;
    auto &r = relays[relay_index];
    if (ee) {
        LOG_TRACE(log, "failed to connect to relay {}: {}", r->device_id.get_short(), ee->message());
        relays[relay_index].reset();
        return connect_to_relay();
    }
    LOG_DEBUG(log, "connected to relay {}", r->device_id.get_short());
    auto &p = res.payload.res;
    master = std::move(p.transport);
    master_endpoint = std::move(p.remote_endpoint);
    rx_buff.resize(BUFF_SZ);
    read_master();
    rx_state |= rx_state_t::response;
    respawn_rx_timer();

    auto tx = utils::bytes_t{};
    proto::relay::serialize(proto::relay::join_relay_request_t{}, tx);
    push_master(tx);
}

void relay_actor_t::on_io_error(const sys::error_code &ec, rotor::plugin::resource_id_t resource) noexcept {
    LOG_TRACE(log, "on_io_error: {}", ec.message());
    resources->release(resource);
    if (ec != asio::error::operation_aborted) {
        LOG_WARN(log, "on_io_error: {}", ec.message());
        if (state < r::state_t::SHUTTING_DOWN) {
            do_shutdown(make_error(ec));
        }
    }
}

void relay_actor_t::on_read(std::size_t bytes) noexcept {
    enum process_t { stop = 1 << 0, more = 1 << 1, incomplete = 1 << 2 };

    LOG_TRACE(log, "on_read: {} bytes, data: {}", bytes, spdlog::to_hex(rx_buff.begin(), rx_buff.begin() + bytes));
    resources->release(resource::io_read);
    rx_idx += bytes;
    size_t from = 0;
    auto process_op = process_t::more;
    while (process_op == process_t::more && from < rx_idx) {
        auto start = (unsigned char *)(rx_buff.data() + from);
        auto sz = rx_idx - from;
        auto r = proto::relay::parse({start, sz});
        process_op = std::visit(
            [&](auto &it) -> process_t {
                using T = std::decay_t<decltype(it)>;
                if constexpr (std::is_same_v<T, proto::relay::incomplete_t>) {
                    return process_t::incomplete;
                } else if constexpr (std::is_same_v<T, proto::relay::protocol_error_t>) {
                    LOG_ERROR(log, "protocol error (master)");
                    auto ec = utils::make_error_code(utils::error_code_t::protocol_error);
                    do_shutdown(make_error(ec));
                    return process_t::stop;
                } else if constexpr (std::is_same_v<T, proto::relay::wrapped_message_t>) {
                    from += it.length;
                    auto ok = on(it.message);
                    return ok ? process_t::more : process_t::stop;
                } else {
                    static_assert(always_false_v<T>, "non-exhaustive visitor!");
                }
            },
            r);
    }
    rx_idx -= from;
    if (process_op != process_t::stop) {
        read_master();
    }
}

bool relay_actor_t::on(proto::relay::message_t &msg) noexcept {
    return std::visit(
        [&](auto &it) -> bool {
            using boost::core::demangle;
            using T = std::decay_t<decltype(it)>;
            auto message_name = demangle(typeid(T).name());
            LOG_TRACE(log, "received message: {}", message_name);
            bool err = false;
            bool cancel_rx_timer = false;
            if constexpr (std::is_same_v<T, proto::relay::pong_t>) {
                if (rx_state & rx_state_t::pong) {
                    rx_state = rx_state & ~rx_state_t::pong;
                    cancel_rx_timer = true;
                } else {
                    err = true;
                }
            } else if constexpr (std::is_same_v<T, proto::relay::ping_t>) {
                auto tx = utils::bytes_t{};
                proto::relay::serialize(proto::relay::pong_t{}, tx);
                push_master(tx);
            } else if constexpr (std::is_same_v<T, proto::relay::response_t>) {
                if (rx_state & rx_state_t::response) {
                    cancel_rx_timer = true;
                    rx_state = rx_state & ~rx_state_t::response;
                    err = !on(it);
                } else {
                    err = true;
                }
            } else if constexpr (std::is_same_v<T, proto::relay::session_invitation_t>) {
                if (rx_state & rx_state_t::invitation) {
                    return on(it);
                } else {
                    err = true;
                }
            } else {
                err = true;
            }
            if (err) {
                LOG_ERROR(log, "protocol error (master, unexpected message): {}", message_name);
                auto ec = utils::make_error_code(utils::error_code_t::protocol_error);
                do_shutdown(make_error(ec));
            }
            if (cancel_rx_timer && rx_timer) {
                cancel_timer(*rx_timer);
                rx_timer.reset();
            }
            return !err;
        },
        msg);
}

bool relay_actor_t::on(proto::relay::response_t &res) noexcept {
    LOG_DEBUG(log, "on response code = {}", res.code);
    if (res.code) {
        LOG_WARN(log, "response error, details = {}", res.details);
        auto ec = utils::make_error_code(utils::error_code_t::relay_failure);
        do_shutdown(make_error(ec));
        return false;
    }
    respawn_ping_timer();
    rx_state |= rx_state_t::invitation;
    auto self = cluster->get_device();
    auto uris = self->get_uris();
    uris.emplace_back(relays[relay_index]->uri);
    using namespace model::diff;
    auto diff = model::diff::cluster_diff_ptr_t{};
    diff = new contact::update_contact_t(*cluster, self->device_id(), uris);
    send<model::payload::model_update_t>(coordinator, std::move(diff), this);
    return true;
}

bool relay_actor_t::on(proto::relay::session_invitation_t &msg) noexcept {
    auto diff = model::diff::cluster_diff_ptr_t{};
    auto device_opt = model::device_id_t::from_sha256(msg.from);
    if (!device_opt) {
        LOG_ERROR(log, "not valid device: {}", spdlog::to_hex(msg.from.begin(), msg.from.end()));
        auto ec = utils::make_error_code(utils::error_code_t::invalid_deviceid);
        do_shutdown(make_error(ec));
        return false;
    }

    asio::ip::tcp::endpoint relay_ep;
    if (msg.address.has_value()) {
        sys::error_code ec;
        relay_ep = asio::ip::tcp::endpoint{msg.address.value(), (uint16_t)msg.port};
    } else {
        relay_ep = asio::ip::tcp::endpoint{master_endpoint.address(), (uint16_t)msg.port};
    }

    auto &device_id = device_opt.value();
    if (cluster->get_devices().by_sha256(device_id.get_sha256())) {
        diff = new model::diff::contact::relay_connect_request_t(std::move(device_id), std::move(msg.key),
                                                                 std::move(relay_ep));
    } else {
        auto last_seen = utils::as_seconds(pt::microsec_clock::local_time());
        db::SomeDevice db;
        db::set_name(db, device_id.get_short());
        db::set_last_seen(db, last_seen);
        diff = new model::diff::modify::add_pending_device_t(device_id, db);
        diff->assign_sibling(new model::diff::contact::unknown_connected_t(*cluster, device_id, std::move(db)));
    }

    send<model::payload::model_update_t>(coordinator, std::move(diff), this);
    return true;
}

void relay_actor_t::respawn_ping_timer() noexcept {
    if (ping_timer) {
        cancel_timer(*ping_timer);
    }
    if (state > r::state_t::OPERATIONAL) {
        return;
    }
    ping_timer = start_timer(relays[relay_index]->ping_interval, *this, &relay_actor_t::on_ping_timer);
}

void relay_actor_t::respawn_tx_timer() noexcept {
    if (tx_timer) {
        cancel_timer(*tx_timer);
    }
    if (state > r::state_t::OPERATIONAL) {
        return;
    }
    tx_timer = start_timer(relays[relay_index]->ping_interval, *this, &relay_actor_t::on_tx_timer);
}

void relay_actor_t::respawn_rx_timer() noexcept {
    if (rx_timer) {
        cancel_timer(*rx_timer);
    }
    if (state > r::state_t::OPERATIONAL) {
        return;
    }
    rx_timer = start_timer(relays[relay_index]->ping_interval, *this, &relay_actor_t::on_rx_timer);
}

void relay_actor_t::on_ping_timer(r::request_id_t, bool cancelled) noexcept {
    ping_timer.reset();
    if (!cancelled) {
        send_ping();
    }
}

void relay_actor_t::on_tx_timer(r::request_id_t, bool cancelled) noexcept {
    tx_timer.reset();
    if (!cancelled) {
        auto ec = utils::make_error_code(utils::error_code_t::tx_timeout);
        do_shutdown(make_error(ec));
    }
}

void relay_actor_t::on_rx_timer(r::request_id_t, bool cancelled) noexcept {
    rx_timer.reset();
    if (!cancelled) {
        auto ec = utils::make_error_code(utils::error_code_t::rx_timeout);
        do_shutdown(make_error(ec));
    }
}

void relay_actor_t::send_ping() noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    auto buff = utils::bytes_t{};
    proto::relay::serialize(proto::relay::ping_t{}, buff);
    push_master(std::move(buff));
    rx_state |= rx_state_t::pong;
}