node.h

Go to the documentation of this file.

#pragma once

 
#include "librats/util/rats_export.h"
#include "librats/transport/connection.h"      // ConnectionDelegate
#include "librats/transport/reactor_pool.h"
#include "librats/core/address.h"
#include "librats/peer/peer_table.h"
#include "librats/peer/peer_id.h"
#include "librats/peer/peer_info.h"
#include "librats/security/identity.h"
#include "librats/security/handshaker.h"  // SecurityProvider
#include "librats/node/circuit_service.h"
#include "librats/node/config.h"
#include "librats/node/dial_service.h"
#include "librats/node/dialer.h"
#include "librats/node/nat_status.h"
#include "librats/node/reflexive_probe.h"
#include "librats/node/node_context.h"    // NodeContext, EventBus, ServiceRegistry
#include "librats/peer/peer.h"
#include "librats/node/peer_network.h"
#include "librats/wire/message_router.h"
 
#include <atomic>
#include <condition_variable>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <string_view>
#include <thread>
#include <type_traits>
#include <unordered_map>
#include <vector>
 
namespace librats {
 
class NetworkMonitor;  // util/network_monitor.h — owned via unique_ptr, included in node.cpp
class MessageJson;     // subsystems/message_json.h — reached via json() (json.h stays out of node.h)
 
class RATS_API Node final : public ConnectionDelegate,
                            public PeerNetwork,
                            public DialService,
                            public CircuitService {
public:
    explicit Node(NodeConfig config);
    ~Node() override;
 
    Node(const Node&) = delete;
    Node& operator=(const Node&) = delete;

    template <class T>
    T* add_subsystem(std::unique_ptr<T> subsystem) {
        static_assert(std::is_base_of<Subsystem, T>::value, "T must derive from Subsystem");
        T* raw = subsystem.get();
        subsystems_.push_back(std::move(subsystem));  // upcast to unique_ptr<Subsystem>
        return raw;
    }

    bool start();
    void stop();

    const PeerId& local_id() const noexcept override { return identity_.id; }
    uint16_t      listen_port() const noexcept override { return listen_port_; }

    uint8_t       transports() const noexcept override { return transports_; }

    const std::string& protocol() const noexcept override { return config_.protocol; }
 
    // — node-scoped coordination, shared by subsystems and the app (see NodeContext) —
    //   events()   : fire-and-forget notifications, one→many (host events, …)
    //   services() : targeted synchronous calls by capability interface, one→one
    EventBus&        events()   noexcept { return events_; }
    ServiceRegistry& services() noexcept { return services_; }
 
    // — connections —
    void connect(const Address& address) override;
    void connect(const std::string& host, uint16_t port);

    size_t                  peer_count() const noexcept { return peers_.size(); }
    size_t                  standbys() const;
    std::vector<PeerInfo>   peers() const override { return peers_.snapshot(); }
    std::optional<Peer> peer(const PeerId& id);

    std::vector<Address> observed_addresses() const;

    const NatStatus& nat_status() const noexcept { return nat_status_; }
 
    // — DialService: dial one endpoint over one wire, bypassing the transport race
    //   (see node/dial_service.h; used by hole punching) —
    bool dial_direct(const Address& addr, TransportKind kind,
                     const DialProfile& profile) override;
 
    // — CircuitService: make a relayed byte stream an ordinary peer connection
    //   (see node/circuit_service.h; used by the relay module) —
    std::optional<PeerRoute> adopt_circuit(const PeerId& carrier, std::unique_ptr<Link> link,
                                           ConnRole role, bool connected) override;
    void                     wake_circuit(PeerRoute route, uint32_t events) override;
    void                     close_circuit(PeerRoute route, CloseReason reason) override;
    std::optional<PeerRoute> dial_relayed(const PeerId& carrier, uint64_t cid,
                                          ForwardedCircuit circuit) override;
    std::optional<PeerRoute> expect_relayed(const PeerId& carrier, uint64_t cid,
                                            ForwardedCircuit circuit) override;
    bool                     forward_datagrams(const PeerId& dialer, const PeerId& responder,
                                               uint64_t cid,
                                               std::shared_ptr<std::atomic<uint64_t>> bytes,
                                               uint64_t rate, uint64_t burst) override;
    void                     stop_forwarding(uint64_t cid) override;
 
    // — peer admission limit (0 = unlimited; guards inbound, not our own dials) —
    size_t max_peers() const noexcept { return max_peers_.load(std::memory_order_relaxed); }
    void   set_max_peers(size_t n) noexcept { max_peers_.store(n, std::memory_order_relaxed); }
    bool   peer_limit_reached() const noexcept {
        const size_t cap = max_peers_.load(std::memory_order_relaxed);
        return cap != 0 && peers_.size() >= cap;
    }
 
    // — application messaging (raw bytes on a named channel) —
    bool send(const PeerId& to, std::string_view channel, ByteView payload, SendOptions options = {});
    bool broadcast(std::string_view channel, ByteView payload, SendOptions options = {});

    bool peer_writable(const PeerId& id) const override;
 
    // — events (register before start(); invoked on a reactor thread). Multiple
    //   listeners are supported, so subsystems and the app can both subscribe. —
    void on_peer_connected(PeerNetwork::PeerEventHandler cb) override { peer_connected_.push_back(std::move(cb)); }
    void on_peer_disconnected(PeerNetwork::PeerDisconnectHandler cb) override { peer_disconnected_.push_back(std::move(cb)); }
    void on_dial_failed(PeerNetwork::DialFailedHandler cb) override { dial_failed_.push_back(std::move(cb)); }
    void on_peer_writable(PeerNetwork::PeerEventHandler cb) override { peer_writable_.push_back(std::move(cb)); }
    void on_peer_rerouted(PeerNetwork::PeerEventHandler cb) override { peer_rerouted_.push_back(std::move(cb)); }
    void on_peer_moved(PeerNetwork::PeerEventHandler cb) override { peer_moved_.push_back(std::move(cb)); }
    void on_peer_path(PeerNetwork::PeerPathHandler cb) override { peer_path_.push_back(std::move(cb)); }
    void on_live_outcome(MessageType type, PeerNetwork::LiveOutcomeHandler cb) override {
        live_outcome_.emplace_back(type, std::move(cb));
    }

    std::optional<PathStats> path_stats(const PeerId& id) const override { return peers_.path(id); }
    void set_send_ceiling(const PeerId& id, uint64_t bits_per_second) override;
    void on(std::string_view channel, MessageRouter::Handler cb) { router_.on_channel(channel, std::move(cb)); }
 
    // — typed lookup of an attached subsystem (nullptr if none of that type) —
    //   reaches a module's own API without threading a pointer from add_subsystem:
    //     if (auto* j = node.json()) j->on("chat", …);
    template <class T>
    T* subsystem() noexcept {
        for (auto& s : subsystems_) if (auto* p = dynamic_cast<T*>(s.get())) return p;
        return nullptr;
    }

    MessageJson* json() noexcept;
 
    // — PeerNetwork (for subsystems) —
    bool                send(const PeerId& to, MessageType type, ByteView payload,
                             SendOptions options = {}) override;
    bool                broadcast(MessageType type, ByteView payload, SendOptions options = {}) override;
    std::vector<PeerId> connected_peers() const override;
    void                on(MessageType type, PeerNetwork::MessageHandler cb) override { router_.on_type(type, std::move(cb)); }
    void                on_with_deadline(MessageType type, PeerNetwork::DeadlineMessageHandler cb) override {
        router_.on_type(type, std::move(cb));
    }
 
private:
    friend class Peer;
 
    // ConnectionDelegate (reactor thread)
    bool admit_inbound() override;
    void on_established(Connection& conn) override;
    void on_frame(Connection& conn, const Frame& frame) override;
    void on_closed(Connection& conn, CloseReason reason) override;
    void on_writable_changed(Connection& conn, bool writable) override;
    void on_path_sample(Connection& conn, const PathStats& stats) override;
    void on_moved(Connection& conn) override;
    void on_live_outcome(Connection& conn, const LiveOutcome& outcome) override;
    void on_dial_aborted(uint8_t reactor_index, ConnId id,
                         const std::string& host, uint16_t port) override;

    bool open_listeners();
    void report_dial_failed(const std::string& host, uint16_t port);
    std::optional<PeerRoute> datagram_route(const PeerId& id) const;
 
    Peer make_peer(const PeerId& id, PeerRoute route) { return Peer(id, route, *this); }
    void route_send(PeerRoute route, FrameHeader header, Bytes payload, Deadline deadline,
                    uint64_t tag, uint8_t rank, std::shared_ptr<std::atomic<size_t>> owed);
    bool send_to(const PeerId& to, FrameHeader header, ByteView payload, SendOptions options);
    bool broadcast_all(FrameHeader header, ByteView payload, SendOptions options);
    size_t send_low_water() const noexcept;
    void route_close(PeerRoute route);
    void close_duplicate(PeerRoute route);
    void take_up(Connection& conn);
 
    // — standby: a duplicate held in case the link it lost to has gone dead —
    void hold_standby(const PeerInfo& info, PeerRoute route);
    void watch_standby(const PeerId& id, uint64_t watch);
    void check_standby(const PeerId& id, uint8_t reactor, uint64_t watch);
    bool take_over(const PeerId& id, PeerRoute from, bool close_from);
 
    // — identify: how peers learn each other's dialable addresses (reactor thread) —
    void                 send_identify(Connection& conn);            
    void                 handle_identify(Connection& conn, const Frame& frame);  
    std::vector<Address> advertised_addresses() const;              
    void                 rebuild_advertised_addresses(const std::vector<std::string>& local_ips);
    void                 record_observed_address(const Address& addr);
    bool                 is_own_endpoint(const Address& addr) const;
 
    void start_network_monitor();   
    void stop_network_monitor();    
    void maintenance_loop();        
 
    NodeConfig                        config_;
    Identity                          identity_;
    std::unique_ptr<SecurityProvider> security_;
    PeerTable                     peers_;
    MessageRouter                     router_;
    EventBus                          events_;      
    ServiceRegistry                   services_;    
    std::unique_ptr<ReactorPool>      reactors_;
    std::unique_ptr<Dialer>           dialer_;      
 
    std::vector<std::unique_ptr<Subsystem>> subsystems_;
 
    // Host network-change watch. The monitor signals on its own thread; the
    // maintenance thread does the (possibly blocking) EventBus emit off it, so
    // subscribers may run slow recovery without stalling change detection.
    std::unique_ptr<NetworkMonitor> monitor_;
    std::thread                     maintenance_thread_;
    std::mutex                      maintenance_mutex_;
    std::condition_variable         maintenance_cv_;
    std::vector<std::string>        pending_addresses_;
    bool                            maintenance_pending_ = false;
    bool                            maintenance_stop_    = false;
 
    socket_t            listen_socket_ = RATS_INVALID_SOCKET;  
    socket_t            udp_socket_    = RATS_INVALID_SOCKET;  
    uint16_t            listen_port_   = 0;   
    uint8_t             transports_    = 0;   
    std::atomic<bool>   running_{false};
    std::atomic<size_t> max_peers_{0};  
 
    std::vector<PeerNetwork::PeerEventHandler>      peer_connected_;
    std::vector<PeerNetwork::PeerDisconnectHandler> peer_disconnected_;
    std::vector<PeerNetwork::DialFailedHandler>     dial_failed_;
    std::vector<PeerNetwork::PeerEventHandler>      peer_writable_;
    std::vector<PeerNetwork::PeerEventHandler>      peer_rerouted_;
    std::vector<PeerNetwork::PeerEventHandler>      peer_moved_;
    std::vector<PeerNetwork::PeerPathHandler>       peer_path_;
    std::vector<std::pair<MessageType, PeerNetwork::LiveOutcomeHandler>> live_outcome_;
 
    // Standbys by peer: what a duplicate connection the peer table turned down
    // is, while it is held (hold_standby).
    struct Standby {
        PeerInfo                              info;
        PeerRoute                             route;
        std::chrono::steady_clock::time_point until;
        uint64_t                              watch = 0;  
    };
    mutable std::mutex                                       standby_mutex_;
    std::unordered_map<PeerId, Standby, PeerId::Hash>       standbys_;
    uint64_t                                                 standby_watches_ = 0;
 
    // Send ceilings by peer (set_send_ceiling), for the link that carries it now
    // and any it moves to.
    mutable std::mutex                                       ceilings_mutex_;
    std::unordered_map<PeerId, uint64_t, PeerId::Hash>      ceilings_;
 
    // Our own addresses as peers observe us (their reported IP + our listen port).
    mutable std::mutex   observed_mutex_;
    std::vector<Address> observed_addresses_;
 
    // The datagram half of the same knowledge, which is a different thing: what a
    // peer sees on a UDP link is our NAT's mapping of the one shared socket, port
    // included, and several peers' views of it are what say whether that mapping is
    // stable enough to punch through. Published as ExternalAddressService.
    NatStatus            nat_status_;
    // And asked of STUN, from that socket, before any peer has said: published as
    // ReflexiveAddressService.
    ReflexiveProbe       reflexive_;
 
    // The dialable addresses we advertise to peers in identify. Derived from local
    // interfaces (and, in future, promoted observed addresses). Rebuilt once at
    // start() and on NetworkMonitor changes — never re-enumerated per connection,
    // since interface enumeration is a syscall and the send path is hot.
    mutable std::mutex   advertised_mutex_;
    std::vector<Address> advertised_addresses_;
};
 
} // namespace librats