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

#include "resolver.h"
#include "model/cluster.h"
#include "model/folder_info.h"
#include "proto/proto-helpers-bep.h"

namespace syncspirit::model {

static advance_action_t compare_by_version(const file_info_t &remote, const file_info_t &local) noexcept {
    auto &r_v = *remote.get_version();
    auto &l_v = *local.get_version();

    auto &r_best = r_v.get_best();
    auto &l_best = l_v.get_best();
    auto rv = proto::get_value(r_best);
    auto lv = proto::get_value(l_best);
    auto r_id = proto::get_id(r_best);
    auto l_id = proto::get_id(l_best);

    // check possible conflict
    if (r_id == l_id) {
        if (lv > rv) {
            return advance_action_t::ignore;
        } else if (lv < rv) {
            return advance_action_t::remote_copy;
        } else {
            return advance_action_t::ignore;
        }
    } else {
        auto r_superior = r_v.contains(l_v);
        auto l_superior = l_v.contains(r_v);
        auto concurrent = !r_superior && !l_superior;
        if (concurrent) {
            if (remote.is_deleted()) {
                return advance_action_t::ignore;
            } else if (local.is_deleted()) {
                return advance_action_t::remote_copy;
            }
        }
        if (r_superior) {
            return advance_action_t::remote_copy;
        }
        if (l_superior) {
            return advance_action_t::ignore;
        }

        auto rm = remote.get_modified_s();
        auto lm = local.get_modified_s();

        if (rm > lm) {
            return advance_action_t::resolve_remote_win;
        } else if (lm > rm) {
            return advance_action_t::ignore;
        }

        return r_id >= l_id ? advance_action_t::resolve_remote_win : advance_action_t::ignore;
    }
}

int compare(const file_info_t &file_1, const file_info_t &file_2) noexcept {
    if (file_1.is_deleted() && file_2.is_deleted()) {
        return 0;
    }
    auto action = compare_by_version(file_2, file_1);
    if (action == advance_action_t::ignore) {
        return 1;
    }
    return -1;
}

static advance_action_t resolve(const file_info_t &remote, const file_info_t *local) noexcept {
    if (remote.is_unreachable()) {
        return advance_action_t::ignore;
    }
    if (remote.is_invalid()) {
        return advance_action_t::ignore;
    }

    auto remote_fi = remote.get_folder_info();
    auto folder = remote_fi->get_folder();
    auto &self = *folder->get_cluster()->get_device();
    auto &folder_infos = folder->get_folder_infos();
    auto &r_v = *remote.get_version();

    for (auto it : folder_infos) {
        auto fi = it.item.get();
        if (fi == remote_fi) {
            continue;
        }
        if (fi->get_device() == &self) {
            continue;
        }
        auto other_party_file = fi->get_file_infos().by_name(remote.get_name());
        if (other_party_file) {
            auto o_v = other_party_file->get_version();
            if (!r_v.contains(*o_v)) {
                return advance_action_t::ignore;
            }
        }
    }
    if (remote.is_deleted() && folder->is_deletion_ignored()) {
        return advance_action_t::ignore;
    }
    if (!local) {
        return advance_action_t::remote_copy;
    }
    // has not been scanned yet, should be re-checked later
    if (!local->is_local()) {
        return advance_action_t::ignore;
    }
    if (remote.is_deleted() && local->is_deleted()) {
        return advance_action_t::ignore;
    }

    return compare_by_version(remote, *local);
}

advance_action_t resolve(const file_info_t &remote) noexcept {
    auto folder = remote.get_folder_info()->get_folder();
    auto self = folder->get_cluster()->get_device();
    auto local_folder = folder->get_folder_infos().by_device(*self);
    auto &local_files = local_folder->get_file_infos();
    auto local_file = local_files.by_name(remote.get_name());
    auto action = resolve(remote, local_file.get());
    if (action == advance_action_t::resolve_remote_win) {
        auto name = remote.get_path().filename().string();
        if (name.find(".sync-conflict-") != std::string::npos) {
            action = advance_action_t::ignore;
        } else {
            auto resolved_name = local_file->make_conflicting_name();
            if (auto resolved = local_files.by_name(resolved_name); resolved) {
                action = advance_action_t::ignore;
            }
        }
    }
    return action;
}

} // namespace syncspirit::model


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

#pragma once

#include "model/file_info.h"
#include "syncspirit-export.h"

namespace syncspirit::model {

enum class advance_action_t {
    ignore = 0,
    remote_copy,
    local_update,
    resolve_remote_win,
};

int SYNCSPIRIT_API compare(const file_info_t &file_1, const file_info_t &file_2) noexcept;

advance_action_t SYNCSPIRIT_API resolve(const file_info_t &remote) noexcept;

} // namespace syncspirit::model

repo_name: basiliscos/syncspirit

license: mit


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

#include "sequencer.h"

namespace syncspirit::model {

sequencer_t::sequencer_t(size_t seed) noexcept : uuid_generator(rng_engine) { rng_engine.seed(seed); }

bu::uuid sequencer_t::next_uuid() noexcept {
    auto lock = lock_t(mutex);
    return uuid_generator();
}

uint64_t sequencer_t::next_uint64() noexcept {
    auto lock = lock_t(mutex);
    return uint64_generator(rng_engine);
}

auto make_sequencer(size_t seed) -> sequencer_ptr_t { return new sequencer_t(seed); }

} // namespace syncspirit::model


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

#pragma once

#include "arc.hpp"
#include "uuid.h"

#include <random>
#include <mutex>
#include <boost/uuid/random_generator.hpp>

namespace syncspirit::model {

using rng_engine_t = std::mt19937;
using uuid_generator_t = boost::uuids::basic_random_generator<rng_engine_t>;
using uint64_generator_t = std::uniform_int_distribution<uint64_t>;

struct SYNCSPIRIT_API sequencer_t : arc_base_t<sequencer_t> {
    sequencer_t(size_t seed) noexcept;
    sequencer_t(sequencer_t &&) = delete;
    sequencer_t(const sequencer_t &) = delete;

    bu::uuid next_uuid() noexcept;
    uint64_t next_uint64() noexcept;

  private:
    using mutex_t = std::mutex;
    using lock_t = std::unique_lock<mutex_t>;
    mutex_t mutex;
    rng_engine_t rng_engine;
    uuid_generator_t uuid_generator;
    uint64_generator_t uint64_generator;
};

using sequencer_ptr_t = intrusive_ptr_t<sequencer_t>;

auto SYNCSPIRIT_API make_sequencer(size_t seed) -> sequencer_ptr_t;

} // namespace syncspirit::model


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

#include "updates_streamer.h"
#include "model/remote_folder_info.h"
#include <algorithm>
#include <spdlog/spdlog.h>

using namespace syncspirit::model;

updates_streamer_t::updates_streamer_t(cluster_t &cluster_, device_t &device) noexcept
    : cluster{cluster_}, peer{&device}, self{cluster.get_device()} {
    refresh_remote();
}

void updates_streamer_t::refresh_remote() noexcept {
    auto &folders = cluster.get_folders();
    auto &remote_folders = peer->get_remote_folder_infos();
    auto streaming_folder = (model::folder_info_t *)(nullptr);
    auto prev_seen = std::move(seen_info);
    seen_info = {};
    for (auto &it : folders) {
        auto &folder = *it.item;
        if (folder.get_folder_type() != db::FolderType::receive) {
            auto peer_folder = folder.is_shared_with(*peer);
            if (peer_folder) {
                auto local_folder = folder.get_folder_infos().by_device(*self);
                if (streaming && streaming->folder_info == local_folder) {
                    streaming_folder = local_folder.get();
                }
                auto remote_folder = remote_folders.by_folder(folder);
                if (remote_folder) {
                    auto seen_sequence = std::int64_t{0};
                    if (remote_folder->get_index() == local_folder->get_index()) {
                        auto previously_seen = prev_seen[local_folder];
                        seen_sequence = std::max(remote_folder->get_max_sequence(), previously_seen);
                    }
                    seen_info[local_folder] = seen_sequence;
                }
            }
        }
    }
    if (streaming && !streaming_folder) {
        streaming.reset();
    } else if (streaming_folder) {
        auto remote_folder = remote_folders.by_folder(*streaming_folder->get_folder());
        if (!remote_folder) {
            streaming.reset();
        } else {
            // TODO, refresh files?
        }
    }
}

bool updates_streamer_t::on_update(file_info_t &file) noexcept {
    assert(file.get_folder_info()->get_device() == self);
    bool r = false;
    auto fi = file.get_folder_info();
    if (!streaming) {
        for (auto &[folder_info, seen_sequence] : seen_info) {
            if (folder_info.get() == fi) {
                auto max = folder_info->get_max_sequence();
                auto unseen_files = file_infos_map_t();
                auto [it, end] = folder_info->get_file_infos().range(seen_sequence + 1, max);
                for (; it != end; ++it) {
                    auto &f = it->item;
                    unseen_files.put(f.get());
                }
                streaming = streaming_info_t(folder_info, std::move(unseen_files));
                r = true;
                break;
            }
        }
    } else if (streaming) {
        auto &info = *streaming;
        if (info.folder_info.get() == file.get_folder_info()) {
            info.unseen_files.put(&file);
            r = true;
        }
    }
    return r;
}

void updates_streamer_t::on_remote_refresh() noexcept { refresh_remote(); }

file_info_ptr_t updates_streamer_t::next() noexcept {
    if (streaming) {
        auto &files = streaming->unseen_files;
        auto &proj = files.sequence_projection();
        if (!proj.empty()) {
            auto it = proj.begin();
            auto file = it->item;
            files.remove(file);
            seen_info[streaming->folder_info] = file->get_sequence();
            return file;
        }
        streaming.reset();
    }
    for (auto &[folder_info, seen_sequence] : seen_info) {
        auto max = folder_info->get_max_sequence();
        if (seen_sequence < max) {
            auto [it, end] = folder_info->get_file_infos().range(seen_sequence + 1, max);
            if (it != end) {
                auto file = it->item;
                seen_info[folder_info] = file->get_sequence();
                ++it;
                if (it != end) {
                    auto unseen_files = file_infos_map_t();
                    for (; it != end; ++it) {
                        auto &f = it->item;
                        unseen_files.put(f.get());
                    }
                    streaming = streaming_info_t(folder_info, std::move(unseen_files));
                }
                return file;
            }
        }
    }
    return {};
}


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

#pragma once

#include "model/cluster.h"
#include "model/file_info.h"
#include "model/device.h"
#include "model/folder_info.h"
#include "model/remote_folder_info.h"
#include "syncspirit-export.h"
#include <optional>
#include <unordered_map>

namespace syncspirit::model {

struct SYNCSPIRIT_API updates_streamer_t {
    updates_streamer_t(cluster_t &, device_t &) noexcept;
    updates_streamer_t(const updates_streamer_t &) noexcept = delete;
    updates_streamer_t(updates_streamer_t &&) noexcept = delete;

    file_info_ptr_t next() noexcept;

    bool on_update(file_info_t &) noexcept;
    void on_remote_refresh() noexcept;
    void on_upsert(const folder_info_t &) noexcept;

  private:
    struct streaming_info_t {
        folder_info_ptr_t folder_info;
        file_infos_map_t unseen_files;
    };
    using streaming_option_t = std::optional<streaming_info_t>;
    using seen_info_t = std::unordered_map<folder_info_ptr_t, std::int64_t>;
    void refresh_remote() noexcept;

    cluster_t &cluster;
    device_ptr_t peer;
    device_ptr_t self;
    seen_info_t seen_info;
    streaming_option_t streaming;
};

} // namespace syncspirit::model


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

#pragma once

#include <string>
#include "../../utils/uri.h"

namespace syncspirit::model {

struct discovery_result {
    utils::uri_ptr_t location;
    std::string search_target;
    std::string usn;
};

struct igd_result {
    std::string control_path;
    std::string description_path;
};

} // namespace syncspirit::model


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

#include "uuid.h"
#include <cassert>

namespace syncspirit::model {

void assign(bu::uuid &uuid, utils::bytes_view_t source) noexcept {
    assert(source.size() == uuid.size());
    std::copy(source.begin(), source.end(), &uuid.data[0]);
}

} // namespace syncspirit::model


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

#pragma once

#include <utils/bytes.h>
#include <boost/uuid/uuid.hpp>
#include "syncspirit-export.h"

namespace syncspirit::model {

namespace bu = boost::uuids;

static const constexpr size_t uuid_length = 16;

SYNCSPIRIT_API void assign(bu::uuid &, utils::bytes_view_t source) noexcept;

} // namespace syncspirit::model


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

#include "pending_device.h"

namespace syncspirit::model {

template <> SYNCSPIRIT_API utils::bytes_view_t get_index<0>(const pending_device_ptr_t &item) noexcept {
    return item->get_device_id().get_sha256();
}

} // namespace syncspirit::model


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

#pragma once

#include "some_device.h"
#include "db/prefix.h"

namespace syncspirit::model {

using pending_device_t = some_device_t<(char)db::prefix::pending_device>;
using pending_devices_map_t = typename pending_device_t::map_t;
using pending_device_ptr_t = typename pending_device_t::ptr_t;

}; // namespace syncspirit::model


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

#include "pending_folder.h"
#include "db/prefix.h"
#include "misc/error_code.h"
#include "proto/proto-helpers.h"

namespace syncspirit::model {

static const constexpr char prefix = (char)(syncspirit::db::prefix::pending_folder);

outcome::result<pending_folder_ptr_t> pending_folder_t::create(utils::bytes_view_t key,
                                                               const db::PendingFolder &data) noexcept {
    if (key.size() != data_length) {
        return make_error_code(error_code_t::invalid_pending_folder_length);
    }

    if (key[0] != prefix) {
        return make_error_code(error_code_t::invalid_folder_prefix);
    }

    auto sha256 = key.subspan(uuid_length + 1);
    auto device = device_id_t::from_sha256(sha256);
    if (!device) {
        return make_error_code(error_code_t::malformed_deviceid);
    }

    auto ptr = pending_folder_ptr_t();
    ptr = new pending_folder_t(key, device.value());
    ptr->assign_fields(data);
    return outcome::success(std::move(ptr));
}

outcome::result<pending_folder_ptr_t> pending_folder_t::create(const bu::uuid &uuid, const db::PendingFolder &data,
                                                               const device_id_t &device_) noexcept {
    auto ptr = pending_folder_ptr_t();
    ptr = new pending_folder_t(uuid, device_);
    ptr->assign_fields(data);
    return outcome::success(std::move(ptr));
}

pending_folder_t::pending_folder_t(const bu::uuid &uuid, const device_id_t &device_) noexcept : device{device_} {
    key[0] = prefix;
    std::copy(uuid.begin(), uuid.end(), key + 1);
    auto sha256 = device.get_sha256();
    std::copy(sha256.begin(), sha256.end(), key + 1 + uuid_length);
}

pending_folder_t::pending_folder_t(utils::bytes_view_t key_, const device_id_t &device_) noexcept : device{device_} {
    std::copy(key_.begin(), key_.end(), key);
}

void pending_folder_t::assign_fields(const db::PendingFolder &data) noexcept {
    auto &f = db::get_folder(data);
    id = db::get_id(f);
    folder_data_t::assign_fields(f);
    auto &fi = db::get_folder_info(data);
    index = db::get_index_id(fi);
    max_sequence = db::get_max_sequence(fi);
}

void pending_folder_t::serialize(db::PendingFolder &data) const noexcept {
    auto &folder = db::get_folder(data);
    folder_data_t::serialize(folder);
    auto &fi = db::get_folder_info(data);
    db::set_index_id(fi, index);
    db::set_max_sequence(fi, max_sequence);
}

utils::bytes_t pending_folder_t::serialize() const noexcept {
    db::PendingFolder r;
    serialize(r);
    return db::encode(r);
}

template <> SYNCSPIRIT_API utils::bytes_view_t get_index<0>(const pending_folder_ptr_t &item) noexcept {
    return item->get_key();
}

template <> SYNCSPIRIT_API utils::bytes_view_t get_index<1>(const pending_folder_ptr_t &item) noexcept {
    auto id = item->get_id();
    auto ptr = (unsigned char *)id.data();
    return {ptr, id.size()};
}

pending_folder_ptr_t pending_folder_map_t::by_key(utils::bytes_view_t key) const noexcept { return get<0>(key); }

pending_folder_ptr_t pending_folder_map_t::by_id(std::string_view id) const noexcept {
    auto ptr = (unsigned char *)id.data();
    auto view = utils::bytes_view_t(ptr, id.size());
    return get<1>(view);
}

} // namespace syncspirit::model


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

#pragma once

#include "misc/augmentation.h"
#include "misc/map.hpp"
#include "misc/uuid.h"
#include "device_id.h"
#include "folder_data.h"
#include "proto/proto-fwd.hpp"
#include "syncspirit-export.h"
#include <boost/outcome.hpp>

namespace syncspirit::model {

namespace outcome = boost::outcome_v2;

struct pending_folder_t;

using pending_folder_ptr_t = intrusive_ptr_t<pending_folder_t>;

struct SYNCSPIRIT_API pending_folder_t final : augmentable_t, folder_data_t {

    static outcome::result<pending_folder_ptr_t> create(utils::bytes_view_t key,
                                                        const db::PendingFolder &data) noexcept;
    static outcome::result<pending_folder_ptr_t> create(const bu::uuid &uuid, const db::PendingFolder &data,
                                                        const device_id_t &device_) noexcept;

    inline const device_id_t &device_id() const noexcept { return device; }
    inline std::uint64_t get_index() const noexcept { return index; }
    inline std::int64_t get_max_sequence() const noexcept { return max_sequence; }
    utils::bytes_view_t get_key() const noexcept { return {key, data_length}; }

    void serialize(db::PendingFolder &data) const noexcept;
    utils::bytes_t serialize() const noexcept;

  private:
    pending_folder_t(utils::bytes_view_t key, const device_id_t &device_) noexcept;
    pending_folder_t(const bu::uuid &uuid, const device_id_t &device_) noexcept;
    void assign_fields(const db::PendingFolder &data) noexcept;

    device_id_t device;
    std::uint64_t index;
    std::int64_t max_sequence;

    static const constexpr auto data_length = uuid_length + device_id_t::digest_length + 1;
    unsigned char key[data_length];
};

struct SYNCSPIRIT_API pending_folder_map_t : generic_map_t<pending_folder_ptr_t, 2> {
    pending_folder_ptr_t by_key(utils::bytes_view_t id) const noexcept;
    pending_folder_ptr_t by_id(std::string_view id) const noexcept;
};

} // namespace syncspirit::model


code:
#include "remote_folder_info.h"
#include "folder.h"
#include "cluster.h"

namespace syncspirit::model {

auto remote_folder_info_t::create(std::uint64_t index_id, std::int64_t max_sequence, device_t &device_,
                                  folder_t &folder_) noexcept -> outcome::result<remote_folder_info_t_ptr_t> {
    auto ptr = remote_folder_info_t_ptr_t{new remote_folder_info_t(index_id, max_sequence, device_, folder_)};
    return outcome::success(ptr);
}

remote_folder_info_t::remote_folder_info_t(std::uint64_t index_id_, std::int64_t max_sequence_, device_t &device_,
                                           folder_t &folder_) noexcept
    : index_id{index_id_}, max_sequence{max_sequence_}, device{&device_}, folder{&folder_} {}

std::string_view remote_folder_info_t::get_key() const noexcept { return folder->get_id(); }

remote_folder_info_t_ptr_t remote_folder_infos_map_t::by_folder(const folder_t &folder) const noexcept {
    auto id = folder.get_id();
    auto ptr = (unsigned char *)id.data();
    auto view = utils::bytes_view_t(ptr, id.size());
    return get<0>(view);
}

auto remote_folder_info_t::get_local() const noexcept -> folder_info_ptr_t {
    auto device = folder->get_cluster()->get_device();
    return folder->get_folder_infos().by_device(*device);
}

template <> SYNCSPIRIT_API utils::bytes_view_t get_index<0>(const remote_folder_info_t_ptr_t &item) noexcept {
    auto key = item->get_key();
    auto ptr = (unsigned char *)key.data();
    return {ptr, key.size()};
}

} // namespace syncspirit::model


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

#pragma once

#include <cstdint>
#include "syncspirit-export.h"
#include <boost/outcome.hpp>
#include "misc/augmentation.h"
#include "misc/map.hpp"

namespace syncspirit::model {

namespace outcome = boost::outcome_v2;

struct folder_t;
using folder_ptr_t = intrusive_ptr_t<folder_t>;

struct device_t;
using device_ptr_t = intrusive_ptr_t<device_t>;

struct remote_folder_info_t;
using remote_folder_info_t_ptr_t = intrusive_ptr_t<remote_folder_info_t>;

struct folder_info_t;
using folder_info_ptr_t = intrusive_ptr_t<folder_info_t>;

struct SYNCSPIRIT_API remote_folder_info_t final : augmentable_t {

    static outcome::result<remote_folder_info_t_ptr_t> create(std::uint64_t index_id, std::int64_t max_sequence,
                                                              device_t &device_, folder_t &folder_) noexcept;

    std::string_view get_key() const noexcept;

    inline std::uint64_t get_index() const noexcept { return index_id; }
    inline std::int64_t get_max_sequence() const noexcept { return max_sequence; }

    folder_info_ptr_t get_local() const noexcept;

  private:
    remote_folder_info_t(std::uint64_t index_id, std::int64_t max_sequence, device_t &device_,
                         folder_t &folder_) noexcept;

    std::uint64_t index_id;
    std::int64_t max_sequence;
    device_t *device;
    folder_t *folder;
};

struct SYNCSPIRIT_API remote_folder_infos_map_t : public generic_map_t<remote_folder_info_t_ptr_t, 1> {
    remote_folder_info_t_ptr_t by_folder(const folder_t &folder) const noexcept;
};

} // namespace syncspirit::model


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

#include "some_device.h"
#include "proto/proto-helpers.h"
#include "utils/time.h"

using namespace syncspirit::model;

some_device_base_t::some_device_base_t(char prefix, const device_id_t &device_id_, const db::SomeDevice &db) noexcept
    : device_id{device_id_} {
    auto sha256 = device_id.get_sha256();
    hash[0] = prefix;
    std::copy(sha256.begin(), sha256.end(), hash + 1);
    assign(db);
}

void some_device_base_t::assign(const db::SomeDevice &db) noexcept {
    name = db::get_name(db);
    client_name = db::get_client_name(db);
    client_version = db::get_client_version(db);
    address = db::get_address(db);
    last_seen = pt::from_time_t(db::get_last_seen(db));
}

syncspirit::utils::bytes_t some_device_base_t::serialize() noexcept {
    db::SomeDevice db;
    serialize(db);
    return db::encode(db);
}

void some_device_base_t::serialize(db::SomeDevice &db) const noexcept {
    db::set_name(db, name);
    db::set_client_name(db, client_name);
    db::set_client_version(db, client_version);
    db::set_address(db, address);
    db::set_last_seen(db, utils::as_seconds(last_seen));
}

std::string_view some_device_base_t::get_name() const noexcept { return name; }
std::string_view some_device_base_t::get_client_name() const noexcept { return client_name; }
std::string_view some_device_base_t::get_client_version() const noexcept { return client_version; }
std::string_view some_device_base_t::get_address() const noexcept { return address; }
const pt::ptime &some_device_base_t::get_last_seen() const noexcept { return last_seen; }
void some_device_base_t::set_last_seen(const pt::ptime &value) noexcept { last_seen = value; }


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

#pragma once

#include <boost/asio.hpp>
#include "misc/augmentation.h"
#include "misc/map.hpp"
#include "misc/error_code.h"

#include "device_id.h"
#include "syncspirit-export.h"
#include "proto/proto-fwd.hpp"

#include <boost/outcome.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

namespace syncspirit::model {

namespace outcome = boost::outcome_v2;
namespace pt = boost::posix_time;

struct SYNCSPIRIT_API some_device_base_t {
    static const constexpr size_t digest_length = 32;
    static const constexpr size_t data_length = digest_length + 1;

    some_device_base_t(const some_device_base_t &other) = delete;
    void assign(const db::SomeDevice &db) noexcept;
    utils::bytes_view_t get_key() const noexcept { return utils::bytes_view_t(hash, data_length); }
    const device_id_t &get_device_id() const noexcept { return device_id; }

    std::string_view get_name() const noexcept;
    std::string_view get_client_name() const noexcept;
    std::string_view get_client_version() const noexcept;
    std::string_view get_address() const noexcept;
    const pt::ptime &get_last_seen() const noexcept;
    void set_last_seen(const pt::ptime &value) noexcept;

    utils::bytes_t serialize() noexcept;
    void serialize(db::SomeDevice &db) const noexcept;

  protected:
    some_device_base_t(char prefix, const device_id_t &device_id_, const db::SomeDevice &db) noexcept;

    std::string name;
    std::string client_name;
    std::string client_version;
    std::string address;
    pt::ptime last_seen;
    device_id_t device_id;
    unsigned char hash[data_length];
};

template <char prefix> struct SYNCSPIRIT_API some_device_t final : some_device_base_t, augmentable_t {
    using ptr_t = intrusive_ptr_t<some_device_t>;

    static outcome::result<ptr_t> create(const device_id_t &id, const db::SomeDevice &db) noexcept {
        return ptr_t(new some_device_t(id, db));
    }

    struct SYNCSPIRIT_API map_t : public generic_map_t<ptr_t, 1> {
        using parent_t = generic_map_t<ptr_t, 1>;
        ptr_t by_sha256(utils::bytes_view_t value) const noexcept { return this->template get<0>(value); }
    };

  private:
    some_device_t(const device_id_t &device_id_, const db::SomeDevice &db) noexcept
        : some_device_base_t(prefix, device_id_, db) {}
};

} // namespace syncspirit::model


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

#include "version.h"
#include "device.h"
#include "proto/proto-helpers.h"
#include "utils/time.h"
#include <limits>

static constexpr auto undef = std::numeric_limits<size_t>::max();

using namespace syncspirit::model;

// auto now = r::pt::second_clock::local_time();

version_t::version_t(const proto::Vector &v) noexcept {
    auto counters_sz = proto::get_counters_size(v);
    assert(counters_sz);
    counters.resize(counters_sz);
    auto best = proto::get_counters(v, 0);
    best_index = 0;
    for (int i = 0; i < counters_sz; ++i) {
        auto &c = proto::get_counters(v, i);
        counters[i] = c;
        if (proto::get_value(c) > proto::get_value(best)) {
            best = c;
            best_index = i;
        }
    }
}

version_t::version_t(const device_t &device) noexcept : best_index{undef} {
    counters.reserve(1);
    update(device);
}

auto version_t::as_proto() const noexcept -> proto::Vector {
    proto::Vector v;
    to_proto(v);
    return v;
}

void version_t::to_proto(proto::Vector &v) const noexcept {
    proto::clear_counters(v);
    for (auto &c : counters) {
        proto::add_counters(v, c);
    }
}

void version_t::update(const device_t &device) noexcept {
    using clock_t = utils::pt::second_clock;
    auto id = device.device_id().get_uint();
    auto v = static_cast<std::uint64_t>(utils::as_seconds(clock_t::universal_time()));
    auto counter = (proto::Counter *)(nullptr);
    if (best_index != undef) {
        v = std::max(proto::get_value(counters[best_index]) + 1, v);
        best_index = undef;
        for (size_t i = 0; i < counters.size(); ++i) {
            if (proto::get_id(counters[i]) == id) {
                counter = &counters[i];
                best_index = i;
                break;
            }
        }
    }
    if (best_index == undef) {
        counters.emplace_back(proto::Counter());
        counter = &counters.back();
        proto::set_id(*counter, id);
        best_index = &counters.back() - counters.data();
    }
    proto::set_value(*counter, v);
}

auto version_t::get_best() noexcept -> proto::Counter & {
    assert(best_index != undef);
    return counters[best_index];
}

auto version_t::get_best() const noexcept -> const proto::Counter & {
    assert(best_index != undef);
    return counters[best_index];
}

bool version_t::contains(const version_t &other) noexcept {
    auto &other_best = other.get_best();
    for (size_t i = 0; i < counters.size(); ++i) {
        auto &c = counters[i];
        auto ids_match = proto::get_id(c) == proto::get_id(other_best);
        if (ids_match) {
            return proto::get_value(c) >= proto::get_value(other_best);
        }
    }
    return false;
}

bool version_t::identical_to(const version_t &other) noexcept {
    if (counters.size() == other.counters.size()) {
        auto p1 = counters.data();
        auto p2 = other.counters.data();
        auto end = p1 + counters.size();
        while (p1 != end) {
            auto ids_match = proto::get_id(*p1) == proto::get_id(*p2);
            auto values_match = proto::get_value(*p1) == proto::get_value(*p2);
            if (ids_match && values_match) {
                ++p1;
                ++p2;
                continue;
            }
            return false;
        }
        return true;
    }
    return false;
}

size_t version_t::counters_size() const { return counters.size(); }

auto version_t::get_counter(size_t index) noexcept -> const proto::Counter & {
    assert(index <= counters.size());
    return counters[index];
}

auto version_t::get_counters() noexcept -> const counters_t & { return counters; }


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

#pragma once

#include "misc/arc.hpp"
#include "syncspirit-export.h"
#include "proto/proto-fwd.hpp"
#include <vector>

namespace syncspirit::model {

struct device_t;

struct SYNCSPIRIT_API version_t final : arc_base_t<version_t> {
    using counters_t = std::vector<proto::Counter>;
    version_t(const proto::Vector &) noexcept;
    version_t(const device_t &) noexcept;

    proto::Vector as_proto() const noexcept;
    void to_proto(proto::Vector &) const noexcept;

    proto::Counter &get_best() noexcept;
    const proto::Counter &get_best() const noexcept;
    void update(const device_t &) noexcept;
    const proto::Counter &get_counter(size_t) noexcept;
    bool contains(const version_t &other) noexcept;
    bool identical_to(const version_t &) noexcept;
    size_t counters_size() const;
    const counters_t &get_counters() noexcept;

  private:
    counters_t counters;
    size_t best_index;
};

using version_ptr_t = intrusive_ptr_t<version_t>;

} // namespace syncspirit::model


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

#include "acceptor_actor.h"
#include "names.h"
#include "utils/format.hpp"
#include "utils/error_code.h"
#include "utils/network_interface.h"
#include "utils/format.hpp"
#include "model/messages.h"
#include "model/diff/contact/connect_request.h"
#include "model/diff/contact/update_contact.h"

using namespace syncspirit::net;
using namespace syncspirit::model::diff;

namespace {
namespace resource {
r::plugin::resource_id_t accepting = 0;
}
} // namespace

acceptor_actor_t::acceptor_actor_t(config_t &config)
    : r::actor_base_t{config}, strand{static_cast<ra::supervisor_asio_t *>(config.supervisor)->get_strand()},
      acceptor(strand.context()), peer(strand.context()), cluster{config.cluster} {}

void acceptor_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.acceptor", 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 acceptor_actor_t::on_start() noexcept {

    LOG_TRACE(log, "on_start");
    sys::error_code ec;

    acceptor.open(endpoint.protocol(), ec);
    if (ec) {
        LOG_ERROR(log, "cannot open endpoint ({}) : {}", endpoint, ec.message());
        return do_shutdown(make_error(ec));
    }

    acceptor.bind(endpoint, ec);
    if (ec) {
        LOG_ERROR(log, "cannot bind endpoint ({}) : {}", endpoint, ec.message());
        return do_shutdown(make_error(ec));
    }

    acceptor.listen(asio::socket_base::max_listen_connections, ec);
    if (ec) {
        LOG_ERROR(log, "cannot listen ({}) : {}", endpoint, ec.message());
        return do_shutdown(make_error(ec));
    }

    endpoint = acceptor.local_endpoint(ec);
    if (ec) {
        LOG_ERROR(log, "cannot get local endpoint {}", ec.message());
        return do_shutdown(make_error(ec));
    }

    auto uris = utils::local_interfaces(endpoint, log);
    if (log->level() <= spdlog::level::debug) {
        for (auto &uri : uris) {
            log->debug("accepting on {}", uri);
        }
    }

    auto diff = model::diff::cluster_diff_ptr_t{};
    diff = new contact::update_contact_t(*cluster, cluster->get_device()->device_id(), uris);
    send<model::payload::model_update_t>(coordinator, std::move(diff), this);
    accept_next();
    r::actor_base_t::on_start();
}

void acceptor_actor_t::accept_next() noexcept {
    resources->acquire(resource::accepting);
    auto fwd = ra::forwarder_t(*this, &acceptor_actor_t::on_accept);
    acceptor.async_accept(peer, std::move(fwd));
}

void acceptor_actor_t::shutdown_start() noexcept {
    r::actor_base_t::shutdown_start();
    if (resources->has(resource::accepting)) {
        sys::error_code ec;
        acceptor.cancel(ec);
        if (ec) {
            LOG_ERROR(log, "cannot cancel accepting :: ", ec.message());
        }
    }
}

void acceptor_actor_t::on_accept(const sys::error_code &ec) noexcept {
    resources->release(resource::accepting);
    if (ec) {
        if (ec != asio::error::operation_aborted) {
            LOG_WARN(log, "accepting error :: ", ec.message());
            return do_shutdown(make_error(ec));
        } else {
            shutdown_continue();
        }
        return;
    }
    sys::error_code err;
    auto remote = peer.remote_endpoint(err);
    if (err) {
        LOG_WARN(log, "on_accept, cannot get remote endpoint:: {}", err.message());
        return accept_next();
    }
    LOG_TRACE(log, "on_accept, peer = {}", remote);

    auto diff = model::diff::cluster_diff_ptr_t{};
    diff = new contact::connect_request_t(std::move(peer), remote);
    send<model::payload::model_update_t>(coordinator, std::move(diff), this);
    accept_next();
}


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

#pragma once

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

namespace syncspirit {
namespace net {

struct acceptor_actor_config_t : r::actor_config_t {
    model::cluster_ptr_t cluster;
};

template <typename Actor> struct acceptor_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 &&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 acceptor_actor_t : public r::actor_base_t {
    using config_t = acceptor_actor_config_t;
    template <typename Actor> using config_builder_t = acceptor_actor_config_builder_t<Actor>;

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

  private:
    using tcp_socket_option_t = boost::optional<tcp_socket_t>;

    void accept_next() noexcept;
    void on_accept(const sys::error_code &ec) noexcept;

    utils::logger_t log;
    asio::io_context::strand &strand;
    tcp::endpoint endpoint;
    tcp::acceptor acceptor;
    tcp_socket_t peer;
    r::address_ptr_t coordinator;
    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 "cluster_supervisor.h"
#include "controller_actor.h"
#include "names.h"
#include "utils/error_code.h"
#include "utils/format.hpp"
#include "hasher/hasher_proxy_actor.h"
#include "model/diff/contact/peer_state.h"

using namespace syncspirit::net;

cluster_supervisor_t::cluster_supervisor_t(cluster_supervisor_config_t &config)
    : ra::supervisor_asio_t{config}, config{config.config}, cluster{config.cluster}, sequencer{config.sequencer} {}

void cluster_supervisor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    ra::supervisor_asio_t::configure(plugin);
    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        p.set_identity("net.cluster", false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.discover_name(names::coordinator, coordinator, false).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(&cluster_supervisor_t::on_model_update, coordinator);
            }
        });
    });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &) {
        create_actor<hasher::hasher_proxy_actor_t>()
            .timeout(init_timeout)
            .hasher_threads(config.hasher_threads)
            .name(net::names::hasher_proxy)
            .finish();
    });
}

void cluster_supervisor_t::on_start() noexcept {
    log->trace("on_start");
    ra::supervisor_asio_t::on_start();
}

void cluster_supervisor_t::shutdown_start() noexcept {
    log->trace("shutdown_start");
    ra::supervisor_asio_t::shutdown_start();
}

void cluster_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.visit(*this, nullptr);
    if (!r) {
        auto ee = make_error(r.assume_error());
        do_shutdown(ee);
    }
}

auto cluster_supervisor_t::operator()(const model::diff::contact::peer_state_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (!cluster->is_tainted()) {
        auto peer = cluster->get_devices().by_sha256(diff.peer_id);
        LOG_TRACE(log, "visiting peer_state_t, {}, state: {}, has been online: {}", peer->device_id(), (int)diff.state,
                  diff.has_been_online);
        if (diff.state == model::device_state_t::online) {
            auto &bep = config.bep_config;
            create_actor<controller_actor_t>()
                .cluster(cluster)
                .sequencer(sequencer)
                .timeout(init_timeout * 7 / 9)
                .peer(peer)
                .peer_addr(diff.peer_addr)
                .connection_id(diff.connection_id)
                .blocks_max_requested(bep.blocks_max_requested)
                .advances_per_iteration(bep.advances_per_iteration)
                .outgoing_buffer_max(bep.tx_buff_limit)
                .request_pool(bep.rx_buff_size)
                .request_timeout(pt::milliseconds(bep.request_timeout))
                .default_path(config.default_location)
                .finish();
        }
    }
    return diff.visit_next(*this, custom);
}

void cluster_supervisor_t::on_child_shutdown(actor_base_t *actor) noexcept {
    LOG_TRACE(log, "on_child_shutdown: {}({})", actor->get_identity(), actor->use_count());
    ra::supervisor_asio_t::on_child_shutdown(actor);
    auto &reason = actor->get_shutdown_reason();
    if (state == r::state_t::OPERATIONAL) {
        log->debug("on_child_shutdown, child {} termination: {}", actor->get_identity(), reason->message());
    }
}


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

#pragma once

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

namespace syncspirit {
namespace net {

namespace r = rotor;
namespace ra = r::asio;
namespace bfs = std::filesystem;
namespace outcome = boost::outcome_v2;

struct cluster_supervisor_config_t : ra::supervisor_config_asio_t {
    config::main_t config;
    model::cluster_ptr_t cluster;
    model::sequencer_ptr_t sequencer;
};

template <typename Supervisor>
struct cluster_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 &&config(const config::main_t &value) && noexcept {
        parent_t::config.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));
    }

    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 cluster_supervisor_t : public ra::supervisor_asio_t, private model::diff::cluster_visitor_t {
    using parent_t = ra::supervisor_asio_t;
    using config_t = cluster_supervisor_config_t;
    template <typename Actor> using config_builder_t = cluster_supervisor_config_builder_t<Actor>;

    explicit cluster_supervisor_t(cluster_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;
    void shutdown_start() noexcept override;

  private:
    void on_model_update(model::message::model_update_t &message) noexcept;

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

    utils::logger_t log;
    r::address_ptr_t coordinator;
    config::main_t config;
    model::cluster_ptr_t cluster;
    model::sequencer_ptr_t sequencer;
};

} // namespace net
} // namespace syncspirit


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

#include "controller_actor.h"
#include "names.h"
#include "constants.h"
#include "model/diff/advance/advance.h"
#include "model/diff/contact/peer_state.h"
#include "model/diff/local/synchronization_finish.h"
#include "model/diff/local/synchronization_start.h"
#include "model/diff/modify/append_block.h"
#include "model/diff/modify/add_remote_folder_infos.h"
#include "model/diff/modify/block_ack.h"
#include "model/diff/modify/block_rej.h"
#include "model/diff/modify/clone_block.h"
#include "model/diff/modify/mark_reachable.h"
#include "model/diff/modify/finish_file.h"
#include "model/diff/modify/remove_peer.h"
#include "model/diff/modify/remove_files.h"
#include "model/diff/modify/remove_folder_infos.h"
#include "model/diff/modify/upsert_folder_info.h"
#include "model/diff/modify/upsert_folder.h"
#include "model/diff/peer/cluster_update.h"
#include "model/diff/peer/update_folder.h"
#include "proto/bep_support.h"
#include "utils/error_code.h"
#include "utils/format.hpp"

#include <utility>

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

namespace {
namespace resource {
r::plugin::resource_id_t peer = 0;
r::plugin::resource_id_t hash = 1;
} // namespace resource

struct context_t {
    bool from_self;
    bool cluster_config_sent;
};

} // namespace

using C = controller_actor_t;

C::folder_synchronization_t::folder_synchronization_t(controller_actor_t &controller_,
                                                      model::folder_t &folder_) noexcept
    : controller{controller_}, folder{&folder_}, synchronizing{false} {}

C::folder_synchronization_t::~folder_synchronization_t() {
    if (blocks.size() && folder && !folder->is_suspended()) {
        controller.push(new model::diff::local::synchronization_finish_t(folder->get_id()));
        for (auto &it : blocks) {
            it.second->unlock();
        }
    }
}

void C::folder_synchronization_t::reset() noexcept { folder.reset(); }

void C::folder_synchronization_t::start_fetching(model::block_info_t *block) noexcept {
    block->lock();
    if (blocks.empty() && !synchronizing) {
        start_sync();
    }
    blocks[block->get_hash()] = model::block_info_ptr_t(block);
}

void C::folder_synchronization_t::finish_fetching(utils::bytes_view_t hash) noexcept {
    auto it = blocks.find(hash);
    it->second->unlock();
    blocks.erase(it);
    if (blocks.size() == 0 && synchronizing) {
        finish_sync();
    }
}

void C::folder_synchronization_t::start_sync() noexcept {
    controller.push(new model::diff::local::synchronization_start_t(folder->get_id()));
    synchronizing = true;
}

void C::folder_synchronization_t::finish_sync() noexcept {
    controller.push(new model::diff::local::synchronization_finish_t(folder->get_id()));
    synchronizing = false;
}

controller_actor_t::controller_actor_t(config_t &config)
    : r::actor_base_t{config}, sequencer{std::move(config.sequencer)}, cluster{config.cluster}, peer{config.peer},
      connection_id{config.connection_id}, peer_addr{config.peer_addr}, request_timeout{config.request_timeout},
      rx_blocks_requested{0}, tx_blocks_requested{0}, outgoing_buffer{0},
      outgoing_buffer_max{config.outgoing_buffer_max}, request_pool{config.request_pool},
      blocks_max_requested{config.blocks_max_requested}, advances_per_iteration{config.advances_per_iteration},
      default_path(std::move(config.default_path)) {
    {
        assert(cluster);
        assert(sequencer);
        current_diff = nullptr;
        planned_pulls = 0;
    }
}

void controller_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) {
        std::string id = "net.controller/";
        id += peer->device_id().get_short();
        p.set_identity(id, false);
        log = utils::get_logger(identity);
        open_reading = p.create_address();
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.discover_name(names::fs_actor, fs_addr, false).link();
        p.discover_name(names::hasher_proxy, hasher_proxy, false).link();
        p.discover_name(names::coordinator, coordinator, false).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(&controller_actor_t::on_model_update, coordinator);
            }
        });
    });
    plugin.with_casted<r::plugin::link_client_plugin_t>([&](auto &p) { p.link(peer_addr, false); });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        p.subscribe_actor(&controller_actor_t::on_forward);
        p.subscribe_actor(&controller_actor_t::on_termination);
        p.subscribe_actor(&controller_actor_t::on_block);
        p.subscribe_actor(&controller_actor_t::on_transfer_pop);
        p.subscribe_actor(&controller_actor_t::on_transfer_push);
        p.subscribe_actor(&controller_actor_t::on_validation);
        p.subscribe_actor(&controller_actor_t::on_block_response);
    });
}

void controller_actor_t::on_start() noexcept {
    r::actor_base_t::on_start();
    LOG_TRACE(log, "on_start");
    if (!owns_best_connection()) {
        LOG_DEBUG(log, "there is a better connection to peer than me ({}), shut self down", connection_id);
        return do_shutdown();
    }

    send<payload::start_reading_t>(peer_addr, get_address(), true);
    send_cluster_config();
    resources->acquire(resource::peer);
    LOG_INFO(log, "is online (connection: {})", connection_id);
    file_iterator = peer->create_iterator(*cluster);
}

void controller_actor_t::shutdown_start() noexcept {
    LOG_TRACE(log, "shutdown_start");
    if (peer_addr) {
        send<payload::termination_t>(peer_addr, shutdown_reason);
    }
    r::actor_base_t::shutdown_start();
}

void controller_actor_t::shutdown_finish() noexcept {
    LOG_TRACE(log, "shutdown_finish, blocks_requested = {}", rx_blocks_requested);
    peer->release_iterator(file_iterator);
    file_iterator.reset();
    synchronizing_folders.clear();
    send_diff();
    r::actor_base_t::shutdown_finish();
}

void controller_actor_t::send_cluster_config() noexcept {
    LOG_TRACE(log, "sending cluster config");
    auto cluster_config = cluster->generate(*peer);
    auto bytes = proto::serialize(cluster_config, peer->get_compression());
    outgoing_buffer += static_cast<uint32_t>(bytes.size());
    send<payload::transfer_data_t>(peer_addr, std::move(bytes));
    send_new_indices();
}

void controller_actor_t::send_new_indices() noexcept {
    if (updates_streamer) {
        auto &remote_folders = peer->get_remote_folder_infos();
        for (auto it : cluster->get_folders()) {
            auto &folder = *it.item;
            auto peer_folder = folder.is_shared_with(*peer);
            if (peer_folder) {
                auto local_folder = folder.get_folder_infos().by_device(*cluster->get_device());
                auto remote_folder = remote_folders.by_folder(folder);
                if (remote_folder && remote_folder->get_index() != local_folder->get_index()) {
                    LOG_DEBUG(log, "sending new index for folder '{}' ({})", folder.get_label(), folder.get_id());
                    proto::Index index;
                    proto::set_folder(index, folder.get_id());
                    auto data = proto::serialize(index, peer->get_compression());
                    outgoing_buffer += static_cast<uint32_t>(data.size());
                    send<payload::transfer_data_t>(peer_addr, std::move(data));
                }
            }
        }
    }
}

void controller_actor_t::on_transfer_push(message::transfer_push_t &message) noexcept {
    auto sz = message.payload.bytes;
    outgoing_buffer += sz;
    LOG_TRACE(log, "on_transfer_push, sz = {} (+{})", outgoing_buffer, sz);
}

void controller_actor_t::on_transfer_pop(message::transfer_pop_t &message) noexcept {
    auto sz = message.payload.bytes;
    assert(outgoing_buffer >= sz);
    outgoing_buffer -= sz;
    LOG_TRACE(log, "on_transfer_pop, sz = {} (-{})", outgoing_buffer, sz);
    push_pending();
}

void controller_actor_t::on_termination(message::termination_signal_t &message) noexcept {
    if (resources->has(resource::peer)) {
        resources->release(resource::peer);
        peer_addr.reset();
    }
    auto &ee = message.payload.ee;
    LOG_TRACE(log, "on_termination reason: {}", ee->message());
    do_shutdown(ee);
}

void controller_actor_t::push_pending() noexcept {
    using pair_t = std::pair<model::folder_info_t *, proto::IndexUpdate>;
    using indices_t = std::vector<pair_t>;

    if (!updates_streamer) {
        return;
    }

    auto indices = indices_t{};
    auto get_index = [&](model::file_info_t &file) -> proto::IndexUpdate & {
        auto folder_info = file.get_folder_info();
        for (auto &p : indices) {
            if (p.first == folder_info) {
                return p.second;
            }
        }
        indices.emplace_back(folder_info, proto::IndexUpdate());
        auto &index = indices.back().second;
        proto::set_folder(index, folder_info->get_folder()->get_id());
        return index;
    };

    auto expected_sz = decltype(outgoing_buffer){0};
    while (expected_sz < outgoing_buffer_max - outgoing_buffer) {
        if (auto file_info = updates_streamer->next(); file_info) {
            expected_sz += file_info->expected_meta_size();
            auto &index = get_index(*file_info);
            proto::add_files(index, file_info->as_proto(true));
            LOG_TRACE(log, "pushing index update for: {}, seq = {}", file_info->get_full_name(),
                      file_info->get_sequence());
        } else {
            break;
        }
    }

    for (auto &p : indices) {
        auto &index = p.second;
        if (proto::get_files_size(index) > 0) {
            auto data = proto::serialize(index, peer->get_compression());
            outgoing_buffer += static_cast<uint32_t>(data.size());
            send<payload::transfer_data_t>(peer_addr, std::move(data));
        }
    }
}

void controller_actor_t::push(model::diff::cluster_diff_ptr_t new_diff) noexcept {
    if (current_diff) {
        current_diff = current_diff->assign_sibling(new_diff.get());
    } else {
        diff = std::move(new_diff);
        current_diff = diff.get();
        while (current_diff->sibling) {
            current_diff = current_diff->sibling.get();
        }
    }
}

void controller_actor_t::pull_ready() noexcept { ++planned_pulls; }

void controller_actor_t::send_diff() noexcept {
    if (planned_pulls && !diff) {
        push(new pull_signal_t(this));
        planned_pulls = 0;
    }
    if (diff) {
        send<model::payload::model_update_t>(fs_addr, std::move(diff), this);
        current_diff = nullptr;
    }
}

void controller_actor_t::on_custom(const pull_signal_t &) noexcept { pull_next(); }

void controller_actor_t::pull_next() noexcept {
    LOG_TRACE(log, "pull_next (pull_signal_t), blocks requested = {}", rx_blocks_requested);
    auto advances = std::uint_fast32_t{0};
    auto can_pull_more = [&]() -> bool {
        bool ignore = (rx_blocks_requested >= blocks_max_requested || request_pool < 0) // rx buff is going to be full
                      || (state != r::state_t::OPERATIONAL) // request pool sz = 32505856e are shutting down
                      || !cluster->get_write_requests() || advances > advances_per_iteration;
        return !ignore;
    };

    using file_set_t = std::set<model::file_info_t *>;
    auto seen_files = file_set_t();
OUTER:
    while (can_pull_more()) {
        if (block_iterator) {
            seen_files.emplace(block_iterator->get_source());
            if (*block_iterator) {
                auto file_block = block_iterator->next();
                if (!file_block.block()->is_locked()) {
                    preprocess_block(file_block);
                }
                continue;
            } else {
                // file_iterator->commit_sync(block_iterator->get_source());
                block_iterator.reset();
            }
            continue;
        }
        if (!block_iterator && !synchronizing_files.empty()) {
            for (auto &[key, guard] : synchronizing_files) {
                auto file = guard->file.get();
                if (seen_files.count(file)) {
                    continue;
                }
                auto bi = model::block_iterator_ptr_t();
                bi = new model::blocks_iterator_t(*file);
                if (bi) {
                    block_iterator = bi;
                    goto OUTER;
                }
            }
        }
        if (auto [file, action] = file_iterator->next(); action != model::advance_action_t::ignore) {
            auto in_sync = synchronizing_files.count(file->get_key());
            if (in_sync) {
                continue;
            }
            if (file->is_locally_available()) {
                auto diff = model::diff::advance::advance_t::create(action, *file, *sequencer);
                if (diff) {
                    ++advances;
                    push(std::move(diff));
                    LOG_TRACE(log, "going to advance on file '{}' from folder '{}'", file->get_name(),
                              file->get_folder_info()->get_folder()->get_label());
                }
            } else if (file->get_size()) {
                auto bi = model::block_iterator_ptr_t();
                bi = new model::blocks_iterator_t(*file);
                if (*bi) {
                    block_iterator = bi;
                    synchronizing_files[file->get_key()] = file->guard();
                }
            }

            continue;
        }
        break;
    }
    if (diff) {
        pull_ready();
        send_diff();
    }
}

void controller_actor_t::preprocess_block(model::file_block_t &file_block) noexcept {
    using namespace model::diff;
    auto file = file_block.file();
    assert(file);
    auto block = file_block.block();
    acquire_block(file_block);

    if (file_block.is_locally_available()) {
        LOG_TRACE(log, "cloning locally available block, file = {}, block index = {} / {}", file->get_full_name(),
                  file_block.block_index(), file->get_blocks().size() - 1);
        auto diff = cluster_diff_ptr_t(new modify::clone_block_t(file_block));
        push_block_write(std::move(diff));
    } else {
        auto sz = block->get_size();
        LOG_TRACE(log, "request_block on file '{}'; block index = {} / {}, sz = {}, request pool sz = {}",
                  file->get_full_name(), file_block.block_index(), file->get_blocks().size() - 1, sz, request_pool);
        request<payload::block_request_t>(peer_addr, file, file_block.block_index()).send(request_timeout);
        ++rx_blocks_requested;
        request_pool -= (int64_t)sz;
    }
}

void controller_actor_t::on_forward(message::forwarded_message_t &message) noexcept {
    if (state != r::state_t::OPERATIONAL) {
        return;
    }
    std::visit([this](auto &msg) { on_message(msg); }, message.payload);
}

void controller_actor_t::on_model_update(model::message::model_update_t &message) noexcept {
    auto custom = const_cast<void *>(message.payload.custom);
    auto pulls = std::uint32_t{0};
    if (custom == this) {
        pulls = planned_pulls;
        planned_pulls = 0;
    }
    auto ctx = context_t{custom == this, false};
    LOG_TRACE(log, "on_model_update, planned pulls = {}", pulls);
    auto &diff = *message.payload.diff;
    auto r = diff.visit(*this, &ctx);
    if (!r) {
        auto ee = make_error(r.assume_error());
        return do_shutdown(ee);
    }
    pull_next();
    push_pending();
    send_diff();
}

auto controller_actor_t::operator()(const model::diff::peer::cluster_update_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto ctx = reinterpret_cast<context_t *>(custom);
    if (ctx->from_self) {
        updates_streamer.reset(new model::updates_streamer_t(*cluster, *peer));
        send_new_indices();
    }
    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::peer::update_folder_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto ctx = reinterpret_cast<context_t *>(custom);
    if (ctx->from_self) {
        auto folder = cluster->get_folders().by_id(diff.folder_id);
        auto &files_map = folder->get_folder_infos().by_device(*peer)->get_file_infos();
        for (auto &f : diff.files) {
            auto file = files_map.by_name(proto::get_name(f));
            cancel_sync(file.get());
        }
        pull_ready();
    }
    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::advance::advance_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto folder = cluster->get_folders().by_id(diff.folder_id);
    if (folder && !folder->is_suspended()) {
        auto &self = *cluster->get_device();
        auto &folder_infos = folder->get_folder_infos();
        auto peer_folder = folder_infos.by_device_id(diff.peer_id);
        auto local_file = model::file_info_ptr_t();
        if (peer_folder) {
            auto name = proto::get_name(diff.proto_source);
            auto file = peer_folder->get_file_infos().by_name(name);
            assert(file);
            if (diff.peer_id == peer->device_id().get_sha256()) {
                local_file = file->local_file();
                if (file->get_blocks().size()) {
                    cancel_sync(file.get());
                }
            }
        }
        if (diff.peer_id == self.device_id().get_sha256()) {
            auto local_folder = folder_infos.by_device(self);
            auto name = proto::get_name(diff.proto_local);
            local_file = local_folder->get_file_infos().by_name(name);
            assert(local_file);
        }
        if (local_file) {
            if (updates_streamer) {
                updates_streamer->on_update(*local_file);
            }
            pull_ready();
        }
    }
    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::contact::peer_state_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (diff.state == model::device_state_t::online && diff.peer_id == peer->device_id().get_sha256()) {
        if (!owns_best_connection()) {
            LOG_DEBUG(log, "there is a better connection to peer than me ({}), shut self down", connection_id);
            do_shutdown();
        }
    }
    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::modify::add_remote_folder_infos_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (diff.device_id == peer->device_id().get_sha256()) {
        if (updates_streamer) {
            updates_streamer->on_remote_refresh();
        }
        push_pending();
    }
    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::modify::upsert_folder_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto folder = cluster->get_folders().by_id(db::get_id(diff.db));
    if (folder->is_shared_with(*peer)) {
        updates_streamer->on_remote_refresh();
        pull_ready();
    }
    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::modify::upsert_folder_info_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto process = diff.device_id == peer->device_id().get_sha256();
    if (process) {
        auto ctx = reinterpret_cast<context_t *>(custom);
        if (!ctx->cluster_config_sent) {
            ctx->cluster_config_sent = true;
            send_cluster_config();
        }
        pull_ready();
    }
    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::modify::remove_files_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (diff.device_id == peer->device_id().get_sha256()) {
        auto requesting_file = block_iterator ? block_iterator->get_source() : nullptr;
        for (auto &key : diff.keys) {
            if (requesting_file) {
                if (requesting_file->get_key() == key) {
                    block_iterator.reset();
                    requesting_file = nullptr;
                }
            }
            auto it = synchronizing_files.find(key);
            if (it != synchronizing_files.end()) {
                synchronizing_files.erase(it);
            }
        }
    }

    return diff.visit_next(*this, custom);
}
auto controller_actor_t::operator()(const model::diff::modify::remove_folder_infos_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    for (auto &key : diff.keys) {
        auto decomposed = model::folder_info_t::decompose_key(key);
        auto device_key = decomposed.device_key();
        if (device_key == peer->get_key() || device_key == cluster->get_device()->get_key()) {
            auto ctx = reinterpret_cast<context_t *>(custom);
            if (!ctx->cluster_config_sent) {
                ctx->cluster_config_sent = true;
                send_cluster_config();
            }
            break;
        }
    }

    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::modify::mark_reachable_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto ctx = reinterpret_cast<context_t *>(custom);
    auto &folder_id = diff.folder_id;
    auto &file_name = diff.file_name;
    auto folder = cluster->get_folders().by_id(folder_id);
    auto &folder_infos = folder->get_folder_infos();
    auto device = cluster->get_devices().by_sha256(diff.device_id);
    auto folder_info = folder_infos.by_device(*device);
    auto file = folder_info->get_file_infos().by_name(file_name);
    if (ctx->from_self) {
        cancel_sync(file.get());
        pull_ready();
    }
    if (device == cluster->get_device() && updates_streamer) {
        updates_streamer->on_update(*file);
    }
    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::modify::block_ack_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto folder = cluster->get_folders().by_id(diff.folder_id);
    if (folder) {
        auto folder_info = folder->get_folder_infos().by_device_id(diff.device_id);
        if (folder_info) {
            auto file = folder_info->get_file_infos().by_name(diff.file_name);
            if (file) {
                if (file->is_locally_available()) {
                    LOG_TRACE(log, "on_block_update, finalizing {}", file->get_name());
                    push(new model::diff::modify::finish_file_t(*file));
                }
            }
        }
    }
    release_block(diff.folder_id, diff.block_hash);
    cluster->modify_write_requests(1);
    pull_ready();
    process_block_write();
    send_diff();

    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::modify::block_rej_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    LOG_ERROR(log, "on block rej, not implemented");
    return diff.visit_next(*this, custom);
}

auto controller_actor_t::operator()(const model::diff::modify::remove_peer_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (diff.get_peer_sha256() == peer->device_id().get_sha256()) {
        LOG_DEBUG(log, "on remove_peer_t, initiating self destruction");
        auto ec = utils::make_error_code(utils::error_code_t::peer_has_been_removed);
        auto reason = make_error(ec);
        do_shutdown(reason);
    }
    return diff.visit_next(*this, custom);
}

void controller_actor_t::on_message(proto::ClusterConfig &message) noexcept {
    using diff_t = model::diff::peer::cluster_update_t;
    LOG_DEBUG(log, "on_message (ClusterConfig)");
    auto diff_opt = diff_t::create(default_path, *cluster, *sequencer, *peer, message);
    if (!diff_opt) {
        auto &ec = diff_opt.assume_error();
        LOG_ERROR(log, "error processing message from {} : {}", peer->device_id(), ec.message());
        return do_shutdown(make_error(ec));
    }
    auto &diff = diff_opt.assume_value();
    push(diff.get());
    pull_ready();
    send_diff();
}

void controller_actor_t::on_message(proto::Index &msg) noexcept {
    LOG_DEBUG(log, "on_message (Index)");
    auto diff_opt = model::diff::peer::update_folder_t::create(*cluster, *sequencer, *peer, msg);
    if (!diff_opt) {
        auto &ec = diff_opt.assume_error();
        LOG_ERROR(log, "error processing message from {} : {}", peer->device_id(), ec.message());
        return do_shutdown(make_error(ec));
    }
    auto &diff = diff_opt.assume_value();
    auto folder = cluster->get_folders().by_id(proto::get_folder(msg));
    auto file_count = proto::get_files_size(msg);
    LOG_DEBUG(log, "on_message (Index), folder = {}, files = {}", folder->get_label(), file_count);
    push(diff.get());
    pull_ready();
    send_diff();
}

void controller_actor_t::on_message(proto::IndexUpdate &msg) noexcept {
    LOG_TRACE(log, "on_message (IndexUpdate)");
    auto diff_opt = model::diff::peer::update_folder_t::create(*cluster, *sequencer, *peer, msg);
    if (!diff_opt) {
        auto &ec = diff_opt.assume_error();
        LOG_ERROR(log, "error processing message from {} : {}", peer->device_id(), ec.message());
        return do_shutdown(make_error(ec));
    }
    auto &diff = diff_opt.assume_value();
    auto folder = cluster->get_folders().by_id(proto::get_folder(msg));
    auto file_count = proto::get_files_size(msg);
    LOG_DEBUG(log, "on_message (IndexUpdate), folder = {}, files = {}", folder->get_label(), file_count);
    push(diff.get());
    pull_ready();
    send_diff();
}

void controller_actor_t::on_message(proto::Request &req) noexcept {
    proto::Response res;
    auto code = proto::ErrorCode::NO_BEP_ERROR;

    if (tx_blocks_requested > blocks_max_requested * constants::tx_blocks_max_factor) {
        LOG_WARN(log, "peer requesting too many blocks ({}), rejecting current request", tx_blocks_requested);
        code = proto::ErrorCode::GENERIC;
    } else {
        auto folder_id = proto::get_folder(req);
        auto folder = cluster->get_folders().by_id(folder_id);
        if (!folder) {
            code = proto::ErrorCode::NO_SUCH_FILE;
        } else {
            auto &folder_infos = folder->get_folder_infos();
            if (folder->is_suspended()) {
                LOG_WARN(log, "folder '{}' is suspended for requests", folder->get_label());
                code = proto::ErrorCode::GENERIC;
            } else {
                auto peer_folder = folder_infos.by_device(*peer);
                if (!peer_folder) {
                    code = proto::ErrorCode::NO_SUCH_FILE;
                } else {
                    auto my_folder = folder_infos.by_device(*cluster->get_device());
                    auto name = proto::get_name(req);
                    auto file = my_folder->get_file_infos().by_name(name);
                    if (!file) {
                        code = proto::ErrorCode::NO_SUCH_FILE;
                    } else {
                        if (!file->is_file()) {
                            LOG_WARN(log, "attempt to request non-regular file: {}", file->get_name());
                            code = proto::ErrorCode::GENERIC;
                        }
                    }
                }
            }
        }
    }

    if (code != proto::ErrorCode::NO_BEP_ERROR) {
        proto::set_id(res, proto::get_id(req));
        proto::set_code(res, code);
        auto data = proto::serialize(res, peer->get_compression());
        outgoing_buffer += static_cast<uint32_t>(data.size());
        send<payload::transfer_data_t>(peer_addr, std::move(data));
    } else {
        ++tx_blocks_requested;
        send<fs::payload::block_request_t>(fs_addr, std::move(req), address);
    }
}

void controller_actor_t::on_message(proto::DownloadProgress &) noexcept {
    LOG_ERROR(log, "on_message, not implemented");
}

void controller_actor_t::on_block_response(fs::message::block_response_t &message) noexcept {
    --tx_blocks_requested;
    auto &p = message.payload;
    proto::Response res;
    proto::set_id(res, proto::get_id(p.remote_request));
    if (p.ec) {
        proto::set_code(res, proto::ErrorCode::GENERIC);
    } else {
        proto::set_data(res, std::move(p.data));
    }

    auto data = proto::serialize(res, peer->get_compression());
    outgoing_buffer += static_cast<uint32_t>(data.size());
    send<payload::transfer_data_t>(peer_addr, std::move(data));
}

void controller_actor_t::on_block(message::block_response_t &message) noexcept {
    --rx_blocks_requested;
    auto &ee = message.payload.ee;
    auto &payload = message.payload.req->payload.request_payload;
    auto file_block = payload.get_block(*cluster, *peer);
    auto file = file_block.file();
    bool do_release_block = false;
    bool try_next = true;
    if (!file) {
        LOG_DEBUG(log, "on_block, file '{}' is not longer available in '{}'", payload.file_name, payload.folder_id);
        do_release_block = true;
    } else {
        auto &block = *file_block.block();
        if (ee) {
            do_release_block = true;
            auto &ec = ee->root()->ec;
            if (ec.category() == utils::request_error_code_category()) {
                auto file = file_block.file();
                if (!file->is_unreachable()) {
                    LOG_WARN(log, "can't receive block from file '{}': {}; marking unreachable", file->get_full_name(),
                             ec.message());
                    file->mark_unreachable(true);
                    push(new model::diff::modify::mark_reachable_t(*file, false));
                    cancel_sync(file);
                }
            } else {
                LOG_WARN(log, "can't receive block : {}", ee->message());
                do_shutdown(ee);
                try_next = false;
            }
        } else {
            auto &data = message.payload.res.data;
            auto hash = file_block.block()->get_hash();
            auto hash_bytes = utils::bytes_t(hash.begin(), hash.end());
            request_pool += block.get_size();

            request<hasher::payload::validation_request_t>(hasher_proxy, data, std::move(hash_bytes), &message)
                .send(init_timeout);
            resources->acquire(resource::hash);
        }
    }

    if (try_next) {
        pull_ready();
    }
    if (do_release_block) {
        release_block(payload.folder_id, payload.block_hash);
    }
    send_diff();
}

void controller_actor_t::on_validation(hasher::message::validation_response_t &res) noexcept {
    using namespace model::diff;
    resources->release(resource::hash);
    auto &ee = res.payload.ee;
    auto block_res = (message::block_response_t *)res.payload.req->payload.request_payload->custom.get();
    auto &payload = block_res->payload.req->payload.request_payload;
    auto file_block = payload.get_block(*cluster, *peer);
    auto file = file_block.file();
    bool do_release_block = false;
    bool try_next = false;
    if (!file) {
        LOG_DEBUG(log, "on_block, file '{}' is not longer available in '{}'", payload.file_name, payload.folder_id);
        do_release_block = true;
    } else {
        auto folder = file->get_folder_info()->get_folder();
        if (ee) {
            do_release_block = true;
            LOG_WARN(log, "on_validation failed : {}", ee->message());
            do_shutdown(ee);
        } else {
            if (!res.payload.res.valid) {
                if (!file->is_unreachable()) {
                    auto ec = utils::make_error_code(utils::protocol_error_code_t::digest_mismatch);
                    LOG_WARN(log, "digest mismatch for '{}'; marking reachable", file->get_full_name(), ec.message());
                    file->mark_unreachable(true);
                    push(new model::diff::modify::mark_reachable_t(*file, false));
                }
                do_release_block = true;
                try_next = true;
            } else if (folder->is_suspended()) {
                LOG_WARN(log, "folder '{}' is suspended, no further processing of '{}'", folder->get_id(),
                         file->get_name());
            } else {
                auto &data = block_res->payload.res.data;
                auto index = payload.block_index;
                LOG_TRACE(log, "{}, got block {}, write requests left = {}", file->get_name(), index,
                          cluster->get_write_requests());
                auto diff = cluster_diff_ptr_t(new modify::append_block_t(*file, index, std::move(data)));
                push_block_write(std::move(diff));
            }
        }
    }
    if (do_release_block) {
        release_block(payload.folder_id, payload.block_hash);
        if (file) {
            cancel_sync(file);
        }
    }
    if (try_next) {
        pull_ready();
        send_diff();
    }
}

void controller_actor_t::push_block_write(model::diff::cluster_diff_ptr_t diff) noexcept {
    block_write_queue.emplace_back(std::move(diff));
    process_block_write();
}

void controller_actor_t::process_block_write() noexcept {
    auto requests_left = cluster->get_write_requests();
    auto sent = 0;
    while (requests_left > 0 && !block_write_queue.empty()) {
        auto &diff = block_write_queue.front();
        push(diff.get());
        --requests_left;
        ++sent;
        block_write_queue.pop_front();
    }
    if (sent) {
        send_diff();
        LOG_TRACE(log, "{} block writes sent, requests left = {}", sent, requests_left);
        cluster->modify_write_requests(-sent);
    }
}

auto controller_actor_t::get_sync_info(model::folder_t *folder) noexcept -> folder_synchronization_ptr_t {
    auto it = synchronizing_folders.find(folder);
    if (it == synchronizing_folders.end()) {
        auto info = folder_synchronization_ptr_t(new folder_synchronization_t(*this, *folder));
        synchronizing_folders.emplace(folder, info);
        return info;
    }
    return it->second;
}

auto controller_actor_t::get_sync_info(std::string_view folder_id) noexcept -> folder_synchronization_ptr_t {
    auto predicate = [folder_id](const auto &it) -> bool { return it.first->get_id() == folder_id; };
    auto it = std::find_if(synchronizing_folders.begin(), synchronizing_folders.end(), predicate);
    assert(it != synchronizing_folders.end());
    return it->second;
}

void controller_actor_t::acquire_block(const model::file_block_t &file_block) noexcept {
    auto block = file_block.block();
    auto folder = file_block.file()->get_folder_info()->get_folder();
    get_sync_info(folder)->start_fetching(block);
}

void controller_actor_t::release_block(std::string_view folder_id, utils::bytes_view_t hash) noexcept {
    get_sync_info(folder_id)->finish_fetching(hash);
}

controller_actor_t::pull_signal_t::pull_signal_t(void *controller_) noexcept : controller{controller_} {}

auto controller_actor_t::pull_signal_t::visit(model::diff::cluster_visitor_t &visitor, void *custom) const noexcept
    -> outcome::result<void> {
    auto r = visitor(*this, custom);
    auto ctx = reinterpret_cast<context_t *>(custom);
    if (r && &visitor == controller && ctx->from_self) {
        auto self = static_cast<controller_actor_t *>(&visitor);
        self->on_custom(*this);
    }
    return r;
}

bool controller_actor_t::owns_best_connection() noexcept { return peer->get_connection_id() == connection_id; }

void controller_actor_t::cancel_sync(model::file_info_t *file) noexcept {
    if (block_iterator && block_iterator->get_source() == file) {
        block_iterator.reset();
    }
    auto it = synchronizing_files.find(file->get_key());
    if (it != synchronizing_files.end()) {
        synchronizing_files.erase(it);
    }
}


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

#pragma once

#include "messages.h"
#include "model/messages.h"
#include "model/cluster.h"
#include "model/diff/local/custom.h"
#include "model/diff/cluster_visitor.h"
#include "model/misc/file_iterator.h"
#include "model/misc/block_iterator.h"
#include "model/misc/updates_streamer.h"
#include "model/misc/sequencer.h"
#include "hasher/messages.h"
#include "utils/log.h"
#include "fs/messages.h"

#include <unordered_map>
#include <deque>

namespace syncspirit {
namespace net {

namespace bfs = std::filesystem;
namespace outcome = boost::outcome_v2;

struct controller_actor_config_t : r::actor_config_t {
    int64_t request_pool;
    model::cluster_ptr_t cluster;
    model::sequencer_ptr_t sequencer;
    model::device_ptr_t peer;
    r::address_ptr_t peer_addr;
    std::string connection_id;
    pt::time_duration request_timeout;
    uint32_t blocks_max_requested = 8;
    uint32_t outgoing_buffer_max = 0;
    std::uint32_t advances_per_iteration = 10;
    bfs::path default_path;
};

template <typename Actor> struct controller_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 &&request_pool(int64_t value) && noexcept {
        parent_t::config.request_pool = 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 &&peer(const model::device_ptr_t &value) && noexcept {
        parent_t::config.peer = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

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

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

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

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

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

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

    builder_t &&outgoing_buffer_max(uint32_t value) && noexcept {
        parent_t::config.outgoing_buffer_max = 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));
    }

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

struct SYNCSPIRIT_API controller_actor_t : public r::actor_base_t, private model::diff::cluster_visitor_t {
    using config_t = controller_actor_config_t;
    template <typename Actor> using config_builder_t = controller_actor_config_builder_t<Actor>;

    controller_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 pull_signal_t final : model::diff::local::custom_t {
        pull_signal_t(void *controller) noexcept;
        outcome::result<void> visit(model::diff::cluster_visitor_t &, void *) const noexcept override;
        void *controller;
    };

    using peers_map_t = std::unordered_map<r::address_ptr_t, model::device_ptr_t>;
    using unlink_request_t = r::message::unlink_request_t;
    using unlink_request_ptr_t = r::intrusive_ptr_t<unlink_request_t>;
    using unlink_requests_t = std::vector<unlink_request_ptr_t>;
    using block_write_queue_t = std::deque<model::diff::cluster_diff_ptr_t>;

    struct folder_synchronization_t : model::arc_base_t<folder_synchronization_t> {
        using block_set_t = std::unordered_map<utils::bytes_view_t, model::block_info_ptr_t>;
        folder_synchronization_t(controller_actor_t &controller, model::folder_t &folder) noexcept;
        ~folder_synchronization_t();
        void reset() noexcept;

        void start_fetching(model::block_info_t *) noexcept;
        void finish_fetching(utils::bytes_view_t hash) noexcept;

        void start_sync() noexcept;
        void finish_sync() noexcept;

      private:
        controller_actor_t &controller;
        model::folder_ptr_t folder;
        block_set_t blocks;
        bool synchronizing;
    };
    using folder_synchronization_ptr_t = model::intrusive_ptr_t<folder_synchronization_t>;
    using synchronizing_folders_t = std::unordered_map<model::folder_ptr_t, folder_synchronization_ptr_t>;
    using synchronizing_files_t = std::unordered_map<utils::bytes_view_t, model::file_info_t::guard_ptr_t>;
    using updates_streamer_ptr_t = std::unique_ptr<model::updates_streamer_t>;

    void on_termination(message::termination_signal_t &message) noexcept;
    void on_forward(message::forwarded_message_t &message) noexcept;
    void on_block(message::block_response_t &message) noexcept;
    void on_validation(hasher::message::validation_response_t &res) noexcept;
    void preprocess_block(model::file_block_t &block) noexcept;
    void on_model_update(model::message::model_update_t &message) noexcept;
    void on_transfer_push(message::transfer_push_t &message) noexcept;
    void on_transfer_pop(message::transfer_pop_t &message) noexcept;
    void on_block_response(fs::message::block_response_t &message) noexcept;

    void on_message(proto::ClusterConfig &message) noexcept;
    void on_message(proto::Index &message) noexcept;
    void on_message(proto::IndexUpdate &message) noexcept;
    void on_message(proto::Request &message) noexcept;
    void on_message(proto::DownloadProgress &message) noexcept;

    void on_custom(const pull_signal_t &diff) noexcept;

    void request_block(const model::file_block_t &block) noexcept;
    void pull_next() noexcept;
    void pull_ready() noexcept;
    void push_pending() noexcept;
    void send_cluster_config() noexcept;
    void send_new_indices() noexcept;
    void push_block_write(model::diff::cluster_diff_ptr_t block) noexcept;
    void process_block_write() noexcept;
    bool owns_best_connection() noexcept;

    void push(model::diff::cluster_diff_ptr_t diff) noexcept;
    void send_diff() noexcept;
    void acquire_block(const model::file_block_t &block) noexcept;
    void release_block(std::string_view folder_id, utils::bytes_view_t hash) noexcept;
    folder_synchronization_ptr_t get_sync_info(model::folder_t *folder) noexcept;
    folder_synchronization_ptr_t get_sync_info(std::string_view folder_id) noexcept;
    void cancel_sync(model::file_info_t *) noexcept;

    outcome::result<void> operator()(const model::diff::advance::advance_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::contact::peer_state_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::add_remote_folder_infos_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::block_ack_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::block_rej_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::mark_reachable_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_peer_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_files_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_folder_infos_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::upsert_folder_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::upsert_folder_info_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::peer::cluster_update_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::peer::update_folder_t &, void *) noexcept override;

    model::sequencer_ptr_t sequencer;
    model::cluster_ptr_t cluster;
    model::device_ptr_t peer;
    std::string connection_id;
    r::address_ptr_t coordinator;
    r::address_ptr_t peer_addr;
    r::address_ptr_t hasher_proxy;
    r::address_ptr_t fs_addr;
    r::address_ptr_t open_reading; /* for routing */
    pt::time_duration request_timeout;
    model::ignored_folders_map_t *ignored_folders;
    // generic
    std::uint_fast32_t rx_blocks_requested;
    std::uint_fast32_t tx_blocks_requested;
    uint32_t outgoing_buffer;
    uint32_t outgoing_buffer_max;

    // diff
    model::diff::cluster_diff_ptr_t diff;
    model::diff::cluster_diff_t *current_diff;
    std::uint32_t planned_pulls;

    int64_t request_pool;
    uint32_t blocks_max_kept;
    uint32_t blocks_max_requested;
    uint32_t advances_per_iteration;
    bfs::path default_path;
    updates_streamer_ptr_t updates_streamer;
    utils::logger_t log;
    unlink_requests_t unlink_requests;
    model::file_iterator_ptr_t file_iterator;
    model::block_iterator_ptr_t block_iterator;
    synchronizing_folders_t synchronizing_folders;
    synchronizing_files_t synchronizing_files;
    block_write_queue_t block_write_queue;
};

} // namespace net
} // namespace syncspirit