node/test/mesh_transport_test.cpp

// Copyright (C) 2026, Lux Industries, Inc. All rights reserved.
// SPDX-License-Identifier: BSD-3-Clause-Eco
//
// mesh_transport_test.cpp — what one hostile or dead socket can do to this node.
//
// The transport is the node's whole attack surface before any signature is
// checked: a peer that has done nothing but connect can already make us read,
// buffer and write. Each case here is a way that used to cost us something and
// now costs the peer its seat:
//
//   [1] a frame length larger than this link's frames  → rejected, peer dropped,
//       and — the part that matters — the buffer does NOT grow while the peer
//       keeps sending. A 4 GiB length header followed by a flood used to be an
//       unbounded allocation on the receiving node.
//   [2] a length under the ZAP ceiling but over the vote cap is the same lie in
//       a smaller hat, and is treated the same way.
//   [3] a peer that votes and then hangs up still counts: the frames already
//       reassembled are delivered BEFORE the corpse is evicted.
//   [4] a socket that will not take a write is gone: broadcast drops it instead
//       of writing to it again every round.
//   [5] eviction is per-peer — the rest of the mesh is untouched.

#include "lux/consensus/zap/vote_codec.hpp"
#include "lux/node/mesh_vote_transport.hpp"
#include "lux/zap/wire.hpp"

#include <chrono>
#include <cstdint>
#include <cstdio>
#include <string>
#include <thread>
#include <vector>

#include <fcntl.h>
#include <sys/socket.h>
#include <unistd.h>

using namespace lux::node;
using namespace lux::consensus;

namespace {

int g_fail = 0;
void check(bool ok, const std::string& what) {
    if (!ok) { std::printf("    ASSERT FAILED: %s\n", what.c_str()); ++g_fail; }
}

// A connected pair: `ours` goes to the transport, `theirs` is the peer we play.
struct Link {
    int ours = -1, theirs = -1;
    Link() {
        int fds[2];
        if (::socketpair(AF_UNIX, SOCK_STREAM, 0, fds) != 0) { std::puts("socketpair"); std::exit(2); }
        ours = fds[0];
        theirs = fds[1];
        // The peer end is non-blocking on the FD, not per call. MSG_DONTWAIT is
        // honoured on recv everywhere and on send only on Linux: on macOS the
        // flood below fills the socket buffer and then blocks in send forever,
        // because the transport has already evicted this peer and nothing is
        // draining. The whole suite stopped on the first test, on one platform.
        // `ours` is left blocking, which is the property the reader documents:
        // MSG_DONTWAIT keeps the transport's own fd blocking so writes stay
        // robust.
        const int flags = ::fcntl(theirs, F_GETFL, 0);
        if (flags >= 0) ::fcntl(theirs, F_SETFL, flags | O_NONBLOCK);
    }
    void close_theirs() { if (theirs >= 0) { ::close(theirs); theirs = -1; } }
    ~Link() { close_theirs(); }
};

// Raw bytes onto the peer end. Returns what the kernel took (a full socket
// buffer is normal when we are deliberately flooding).
std::size_t send_raw(int fd, const std::vector<std::uint8_t>& b) {
    const ssize_t w = ::send(fd, b.data(), b.size(), MSG_DONTWAIT);
    return w > 0 ? static_cast<std::size_t>(w) : 0;
}

std::vector<std::uint8_t> header(std::uint32_t len, std::uint8_t type) {
    lux::zap::Writer w;
    w.write_u32(len);
    std::vector<std::uint8_t> b = w.take();
    b.push_back(type);
    return b;
}

// A structurally valid vote frame (decode_vote checks widths, not signatures).
std::vector<std::uint8_t> vote_frame(std::uint8_t tag) {
    SignedVote v;
    v.block_id.fill(tag);
    v.voter.fill(0x01);
    v.sig.fill(0x02);
    const std::vector<std::uint8_t> payload = lux::consensus::zap::encode_vote(v);
    std::vector<std::uint8_t> f = header(static_cast<std::uint32_t>(payload.size()),
                                         lux::consensus::zap::kVoteMsgType);
    f.insert(f.end(), payload.begin(), payload.end());
    return f;
}

}  // namespace

int main() {
    std::printf("=============== node — mesh transport under a hostile peer ===============\n");
    std::printf("one socket must never cost this node unbounded memory, a stalled write, or a seat it kept\n\n");

    std::size_t delivered = 0;
    auto counting_sink = [&delivered](const SignedVote&) { ++delivered; };

    // [1] oversize length + flood: rejected, evicted, and the buffer stays flat.
    {
        MeshVoteTransport tx(counting_sink);
        Link l;
        tx.add_peer(l.ours, 1);
        send_raw(l.theirs, header(0xFFFFFFFFu, lux::consensus::zap::kVoteMsgType));

        // Flood as hard as the peer can while the transport keeps pumping. The
        // old reader latched its error flag and then went on buffering every one
        // of these bytes; nothing consumed the flag, so the node grew without limit.
        std::vector<std::uint8_t> junk(64 * 1024, 'A');
        std::size_t sent = 0;
        for (int i = 0; i < 256; ++i) {
            sent += send_raw(l.theirs, junk);
            tx.pump();
        }
        check(tx.peer_count() == 0, "[1] the peer that announced a 4 GiB frame is gone");
        check(tx.evicted() == 1, "[1] exactly one eviction");
        check(delivered == 0, "[1] no vote was delivered from the flood");
        std::printf("  [1] flood: peer sent %zu KiB, peers left %zu, evictions %zu\n",
                    sent / 1024, tx.peer_count(), tx.evicted());
    }

    // [2] a length the ZAP ceiling would allow but this link never carries. The
    // link's cap is kMaxMeshFrame, not a vote's 188 bytes: the same socket now
    // carries transactions as well, and the reader that enforces the bound sees
    // frames rather than meanings. What matters is that the bound is far below
    // ZAP's 16 MiB and that crossing it costs the peer its place.
    {
        delivered = 0;
        MeshVoteTransport tx(counting_sink);
        Link l;
        tx.add_peer(l.ours, 1);
        const std::uint32_t over = kMaxMeshFrame + 1;
        check(over < lux::zap::MaxMessageSize, "[2] the test length is legal ZAP, illegal here");
        send_raw(l.theirs, header(over, lux::consensus::zap::kVoteMsgType));
        tx.pump();
        check(tx.peer_count() == 0, "[2] a frame too large for this link drops the peer");
    }

    // [2b] a message type this node does not serve is DROPPED, and the peer that
    // sent it keeps its place. Dispatch is by registration, so an unregistered
    // type costs nothing — a peer cannot make this node work by naming one.
    {
        delivered = 0;
        MeshVoteTransport tx(counting_sink);
        Link l;
        tx.add_peer(l.ours, 1);
        std::vector<std::uint8_t> f = header(4, 0x2A);   // a type nothing registered
        for (int i = 0; i < 4; ++i) f.push_back(0xEE);
        send_raw(l.theirs, f);
        tx.pump();
        check(tx.peer_count() == 1, "[2b] an unserved message type does not drop the peer");
        check(delivered == 0, "[2b] and delivers nothing");
    }

    // [2c] a registered type IS delivered, on the same socket votes ride.
    {
        delivered = 0;
        std::size_t got = 0;
        MeshVoteTransport tx(counting_sink);
        tx.on(kTxMsgType, [&](std::uint32_t, const std::vector<std::uint8_t>& p) { got += p.size(); });
        Link l;
        tx.add_peer(l.ours, 1);
        std::vector<std::uint8_t> f = header(3, kTxMsgType);
        for (int i = 0; i < 3; ++i) f.push_back(0x11);
        send_raw(l.theirs, f);
        tx.pump();
        check(got == 3, "[2c] a registered message type reaches its handler");
        check(tx.peer_count() == 1, "[2c] and the peer keeps its place");
    }

    // [3] a validator that votes and then hangs up still counts.
    {
        delivered = 0;
        MeshVoteTransport tx(counting_sink);
        Link l;
        tx.add_peer(l.ours, 1);
        send_raw(l.theirs, vote_frame(0x42));
        l.close_theirs();  // EOF arrives with the vote already in flight
        tx.pump();
        check(delivered == 1, "[3] the vote of a peer that then disconnected was delivered");
        check(tx.peer_count() == 0, "[3] and the closed peer was evicted");
    }

    // [4] a socket that cannot take a write is gone, not retried forever.
    {
        delivered = 0;
        MeshVoteTransport tx(counting_sink);
        Link l;
        tx.add_peer(l.ours, 1);
        l.close_theirs();
        ::shutdown(l.ours, SHUT_RDWR);  // writes now fail outright

        SignedVote v;
        v.block_id.fill(0x43);
        v.voter.fill(0x01);
        v.sig.fill(0x02);
        tx.broadcast(v);
        check(delivered == 1, "[4] broadcast still self-echoes to our own gate");
        // The write fails on the peer's own writer; the driver drops it on the
        // next pump.
        for (int i = 0; i < 200 && tx.peer_count() != 0; ++i) {
            tx.pump();
            std::this_thread::sleep_for(std::chrono::milliseconds(5));
        }
        check(tx.peer_count() == 0, "[4] the unwritable peer was dropped");
    }

    // [4b] a peer that stops reading never holds the thread that sends. Frames
    // for it wait in its own queue — bounded, the oldest let go — and its writer
    // times out and takes the peer with it; the honest peer beside it is sent
    // everything, at once.
    {
        delivered = 0;
        MeshVoteTransport tx(counting_sink);
        Link stuck, honest;
        tx.add_peer(stuck.ours, 1);
        tx.add_peer(honest.ours, 2);
        const std::vector<std::uint8_t> big(512 * 1024, 0x5a);
        const auto t0 = std::chrono::steady_clock::now();
        for (int i = 0; i < 64; ++i) tx.gossip(kTxMsgType, big);  // 32 MiB, far past what a socket buffers
        const auto took = std::chrono::duration_cast<std::chrono::milliseconds>(
                              std::chrono::steady_clock::now() - t0).count();
        // One blocked write would cost kWriteTimeoutMs; 64 of them, two minutes.
        check(took < kWriteTimeoutMs / 4, "[4b] 64 half-MiB frames to a peer that reads nothing returned "
                                          "without waiting on its socket (" + std::to_string(took) + " ms)");
        // The honest peer reads its share while the stuck one reads nothing.
        std::size_t got = 0;
        std::uint8_t buf[65536];
        const auto   until = std::chrono::steady_clock::now() + std::chrono::seconds(10);
        while (got < 4 * (big.size() + 5) && std::chrono::steady_clock::now() < until) {
            const ssize_t n = ::recv(honest.theirs, buf, sizeof buf, MSG_DONTWAIT);
            if (n > 0) got += std::size_t(n);
            else std::this_thread::sleep_for(std::chrono::milliseconds(1));
            tx.pump();
        }
        check(got >= 4 * (big.size() + 5), "[4b] the peer that reads was sent to meanwhile");
        const auto gone = std::chrono::steady_clock::now() + std::chrono::milliseconds(3 * kWriteTimeoutMs);
        while (tx.peer_count() == 2 && std::chrono::steady_clock::now() < gone) {
            (void)::recv(honest.theirs, buf, sizeof buf, MSG_DONTWAIT);
            tx.pump();
            std::this_thread::sleep_for(std::chrono::milliseconds(5));
        }
        check(tx.peer_count() == 1 && tx.members() == std::vector<std::uint32_t>{2},
              "[4b] the peer that stopped reading was dropped when its writer timed out; the other stays");
    }

    // [5] eviction is per-peer: one hostile socket does not cost an honest one.
    {
        delivered = 0;
        MeshVoteTransport tx(counting_sink);
        Link bad, good;
        tx.add_peer(bad.ours, 1);
        tx.add_peer(good.ours, 2);
        send_raw(bad.theirs, header(0xFFFFFFFFu, lux::consensus::zap::kVoteMsgType));
        send_raw(good.theirs, vote_frame(0x44));
        tx.pump();
        check(tx.peer_count() == 1, "[5] the honest peer kept its seat");
        check(delivered == 1, "[5] and its vote arrived");
        send_raw(good.theirs, vote_frame(0x45));
        tx.pump();
        check(delivered == 2, "[5] the honest peer still works after the eviction");
    }

    std::printf("---------------------------------------------------------------------------\n");
    if (g_fail) { std::printf("==== node MESH TRANSPORT: FAIL (%d) ====\n", g_fail); return 1; }
    std::printf("==== node MESH TRANSPORT: PASS — a hostile peer costs itself, not the node ====\n");
    return 0;
}