initial

modules/peer_to_peer/src/PeerAction.hpp

@@ -0,0 +1,15 @@
+#pragma once
+
+#include <cstdint>
+#include <glm/glm.hpp>
+
+namespace tw::p2p {
+
+struct PeerAction {
+    uint32_t  peer_id;
+    uint32_t  frame_idx;
+    glm::vec3 input;
+    uint32_t  ack_frame{0};
+};
+
+}

modules/peer_to_peer/src/PeerLink.cpp

@@ -0,0 +1,86 @@
+#include "PeerLink.hpp"
+
+#include <cstring>
+#include <spdlog/spdlog.h>
+
+namespace tw::p2p {
+
+PeerLink::PeerLink(uint32_t self_id, uint32_t remote_id,
+                   tw::net::quicr::QuicrConnection* connection)
+    : m_self_id(self_id)
+    , m_remote_id(remote_id)
+    , m_connection(connection)
+    , m_recv_buffer(64 * 1024)
+{}
+
+// ── outbound ──────────────────────────────────────────────────────────────────
+
+void PeerLink::send_hello(const mmo::peer::PeerHello& msg) {
+    send(MsgType::Hello, msg, /*reliable=*/true);
+}
+
+void PeerLink::send_action(const mmo::peer::PeerAction& msg) {
+    send(MsgType::Action, msg, /*reliable=*/false);
+}
+
+void PeerLink::send_bye(const mmo::peer::PeerBye& msg) {
+    send(MsgType::Bye, msg, /*reliable=*/true);
+}
+
+void PeerLink::send(MsgType type, const google::protobuf::MessageLite& msg, bool reliable) {
+    std::string payload = msg.SerializeAsString();
+
+    std::vector<std::byte> frame(sizeof(uint32_t) + payload.size());
+    uint32_t tag = static_cast<uint32_t>(type);
+    std::memcpy(frame.data(), &tag, sizeof(tag));
+    std::memcpy(frame.data() + sizeof(tag), payload.data(), payload.size());
+
+    auto r = m_connection->send_message(std::span(frame), reliable);
+    if (!r) {
+        spdlog::warn("PeerLink[{}->{}]: send failed", m_self_id, m_remote_id);
+    }
+}
+
+// ── inbound ───────────────────────────────────────────────────────────────────
+
+void PeerLink::poll() {
+    while (true) {
+        auto r = m_connection->read_into(std::span(m_recv_buffer));
+        if (!r || *r == 0) break;
+        dispatch(std::span<const std::byte>(m_recv_buffer.data(), *r));
+    }
+}
+
+void PeerLink::dispatch(std::span<const std::byte> frame) {
+    if (frame.size() < sizeof(uint32_t)) return;
+
+    uint32_t tag;
+    std::memcpy(&tag, frame.data(), sizeof(tag));
+    auto payload = frame.subspan(sizeof(tag));
+    spdlog::info("Received into a link");
+
+    switch (static_cast<MsgType>(tag)) {
+        case MsgType::Hello: {
+            mmo::peer::PeerHello msg;
+            if (msg.ParseFromArray(payload.data(), static_cast<int>(payload.size())) && m_hello_handler)
+                m_hello_handler(msg);
+            break;
+        }
+        case MsgType::Action: {
+            mmo::peer::PeerAction msg;
+            if (msg.ParseFromArray(payload.data(), static_cast<int>(payload.size())) && m_action_handler)
+                m_action_handler(msg);
+            break;
+        }
+        case MsgType::Bye: {
+            mmo::peer::PeerBye msg;
+            if (msg.ParseFromArray(payload.data(), static_cast<int>(payload.size())) && m_bye_handler)
+                m_bye_handler(msg);
+            break;
+        }
+        default:
+            spdlog::warn("PeerLink[{}]: unknown message type {}", m_self_id, tag);
+    }
+}
+
+}

modules/peer_to_peer/src/PeerLink.hpp

@@ -0,0 +1,42 @@
+#pragma once
+
+#include "PeerAction.hpp"
+#include "Peer.pb.h"
+#include "Address.hpp"
+
+#include <cstdint>
+#include <functional>
+
+namespace tw::p2p {
+
+// Node-level peer-networking abstraction.
+//
+// Owns its transport (QUIC endpoint or TCP listener), manages all peer
+// connections, and internalises the PeerHello/PeerBye handshake.
+//
+// External code only sees:
+//   on_connected(peer_id)   — handshake with peer_id complete (both hellos sent)
+//   on_action(peer_id, a)   — PeerAction received from peer_id
+//   connect_to(peer_id, a)  — initiate outgoing connection
+//   send_action(peer_id, m) — unreliable send to a connected peer
+//   poll()                  — pump I/O; must be called regularly
+class PeerLink {
+public:
+    using ConnectedHandler = std::function<void(uint32_t peer_id)>;
+    using ActionHandler    = std::function<void(uint32_t peer_id, const PeerAction& action)>;
+
+    virtual ~PeerLink() = default;
+
+    void on_connected(ConnectedHandler h) { m_connected_handler = std::move(h); }
+    void on_action   (ActionHandler    h) { m_action_handler    = std::move(h); }
+
+    virtual void connect_to (uint32_t peer_id, const tw::net::Address& addr) = 0;
+    virtual void send_batch (uint32_t peer_id, const mmo::peer::PeerActionBatch& batch) = 0;
+    virtual void poll       () = 0;
+
+protected:
+    ConnectedHandler m_connected_handler;
+    ActionHandler    m_action_handler;
+};
+
+} // namespace tw::p2p

modules/peer_to_peer/src/PeerRegistry.cpp

@@ -0,0 +1,64 @@
+#include "PeerRegistry.hpp"
+
+#include <fcntl.h>
+#include <sys/file.h>
+#include <unistd.h>
+
+#include <sstream>
+#include <stdexcept>
+
+namespace tw::p2p {
+
+PeerRegistry::PeerRegistry(std::string file_path)
+    : m_file_path(std::move(file_path)) {}
+
+void PeerRegistry::register_self(uint32_t id, const std::string& address) {
+    int fd = ::open(m_file_path.c_str(), O_RDWR | O_CREAT, 0644);
+    if (fd < 0) throw std::runtime_error("PeerRegistry: failed to open registry file");
+
+    // Blocks until exclusive lock is granted.
+    if (::flock(fd, LOCK_EX) != 0) {
+        ::close(fd);
+        throw std::runtime_error("PeerRegistry: flock failed");
+    }
+
+    std::string line = std::to_string(id) + "," + address + "\n";
+    ::lseek(fd, 0, SEEK_END);
+    ::write(fd, line.data(), line.size());
+
+    ::flock(fd, LOCK_UN);
+    ::close(fd);
+}
+
+std::vector<PeerRegistry::PeerInfo> PeerRegistry::read_all() const {
+    std::vector<PeerInfo> result;
+
+    int fd = ::open(m_file_path.c_str(), O_RDONLY);
+    if (fd < 0) return result;
+
+    ::flock(fd, LOCK_SH);
+
+    std::string content;
+    char buf[512];
+    ssize_t n;
+    while ((n = ::read(fd, buf, sizeof(buf))) > 0)
+        content.append(buf, static_cast<size_t>(n));
+
+    ::flock(fd, LOCK_UN);
+    ::close(fd);
+
+    std::istringstream stream(content);
+    std::string line;
+    while (std::getline(stream, line)) {
+        if (line.empty()) continue;
+        auto comma = line.find(',');
+        if (comma == std::string::npos) continue;
+        PeerInfo info;
+        info.id      = static_cast<uint32_t>(std::stoul(line.substr(0, comma)));
+        info.address = line.substr(comma + 1);
+        result.push_back(std::move(info));
+    }
+    return result;
+}
+
+}

modules/peer_to_peer/src/PeerRegistry.hpp

@@ -0,0 +1,27 @@
+#pragma once
+
+#include <cstdint>
+#include <string>
+#include <vector>
+
+namespace tw::p2p {
+
+class PeerRegistry {
+public:
+    struct PeerInfo {
+        uint32_t id;
+        std::string address;
+    };
+
+    explicit PeerRegistry(std::string file_path);
+
+    // Blocks until the file lock is acquired, appends this node's entry, then releases.
+    void register_self(uint32_t id, const std::string& address);
+
+    std::vector<PeerInfo> read_all() const;
+
+private:
+    std::string m_file_path;
+};
+
+}

modules/peer_to_peer/src/PeerToPeer.cpp

@@ -0,0 +1,147 @@
+#include "TestPeerNode.hpp"
+#include "QuicrPeerLink.hpp"
+#include "TcpPeerLink.hpp"
+#include "runtime/LockStep.hpp"
+
+#include <atomic>
+#include <csignal>
+#include <iostream>
+#include <memory>
+#include <spdlog/spdlog.h>
+#include <string>
+
+// ── signal handling ───────────────────────────────────────────────────────────
+
+static std::atomic<bool> g_running{true};
+
+static void on_signal(int) {
+    g_running.store(false);
+}
+
+// ── CLI ───────────────────────────────────────────────────────────────────────
+
+enum class Transport { Quicr, Tcp };
+
+struct Config {
+    uint32_t    id           = 0;
+    uint16_t    port         = 0;
+    std::string registry     = "";
+    uint32_t    peer_count   = 0;   // total number of nodes to wait for
+    int         ticks        = 0;   // 0 = run until SIGINT
+    Transport   transport    = Transport::Quicr;
+};
+
+static void print_usage(const char* prog) {
+    std::cerr
+        << "Usage: " << prog
+        << " --id <N> --port <N> --registry <path> --peers <N> [--ticks <N>]\n"
+        << "\n"
+        << "  --id        Unique node ID (uint, > 0)\n"
+        << "  --port      UDP port this node listens on\n"
+        << "  --registry  Path to shared CSV registry file\n"
+        << "  --peers     Total number of nodes that will participate\n"
+        << "  --ticks     How many simulation frames to run (default: run until SIGINT)\n"
+        << "  --transport quicr|tcp  Transport backend (default: quicr)\n";
+}
+
+static bool parse_args(int argc, char* argv[], Config& cfg) {
+    for (int i = 1; i < argc; ++i) {
+        std::string a = argv[i];
+        if (a == "--help" || a == "-h") return false;
+
+        auto need_next = [&]() -> const char* {
+            if (i + 1 >= argc) { std::cerr << a << " requires a value\n"; return nullptr; }
+            return argv[++i];
+        };
+
+        if      (a == "--id")        { auto v = need_next(); if (!v) return false; cfg.id         = std::stoul(v); }
+        else if (a == "--port")      { auto v = need_next(); if (!v) return false; cfg.port       = static_cast<uint16_t>(std::stoul(v)); }
+        else if (a == "--registry")  { auto v = need_next(); if (!v) return false; cfg.registry   = v; }
+        else if (a == "--peers")     { auto v = need_next(); if (!v) return false; cfg.peer_count = std::stoul(v); }
+        else if (a == "--ticks")     { auto v = need_next(); if (!v) return false; cfg.ticks      = std::stoi(v); }
+        else if (a == "--transport") {
+            auto v = need_next(); if (!v) return false;
+            std::string t = v;
+            if      (t == "quicr") cfg.transport = Transport::Quicr;
+            else if (t == "tcp")   cfg.transport = Transport::Tcp;
+            else { std::cerr << "Unknown transport: " << t << " (use quicr or tcp)\n"; return false; }
+        }
+        else { std::cerr << "Unknown argument: " << a << "\n"; return false; }
+    }
+
+    if (cfg.id == 0 || cfg.port == 0 || cfg.registry.empty() || cfg.peer_count == 0) {
+        std::cerr << "Error: --id, --port, --registry, and --peers are all required\n";
+        return false;
+    }
+    return true;
+}
+
+// ── main ──────────────────────────────────────────────────────────────────────
+
+int main(int argc, char* argv[]) {
+    Config cfg;
+    if (!parse_args(argc, argv, cfg)) {
+        print_usage(argv[0]);
+        return 1;
+    }
+
+    std::signal(SIGINT,  on_signal);
+    std::signal(SIGTERM, on_signal);
+
+    spdlog::set_level(spdlog::level::info);
+    spdlog::info("[p2p] node {} starting on port {}", cfg.id, cfg.port);
+
+    // ── setup ─────────────────────────────────────────────────────────────────
+
+    const char* ext = (cfg.transport == Transport::Tcp) ? "tcp" : "quicr";
+    std::string csv_path = "/tmp/peer_node_" + std::to_string(cfg.id) + "." + ext + ".csv";
+
+    std::unique_ptr<tw::p2p::PeerLink> link;
+    if (cfg.transport == Transport::Tcp)
+        link = std::make_unique<tw::p2p::TcpPeerLink>(cfg.id, tw::net::Address{std::nullopt, cfg.port});
+    else
+        link = std::make_unique<tw::p2p::QuicrPeerLink>(cfg.id, cfg.port);
+
+    tw::p2p::TestPeerNode node(cfg.id, cfg.port, std::move(link), cfg.registry, csv_path);
+
+    node.register_self();
+    spdlog::info("[p2p] node {} registered; waiting for {} peers …", cfg.id, cfg.peer_count);
+
+    if (!node.wait_until_peers_registered(cfg.peer_count)) {
+        spdlog::error("[p2p] node {}: timed out waiting for peers — aborting", cfg.id);
+        return 1;
+    }
+
+    spdlog::info("[p2p] node {}: all {} peers registered; connecting …", cfg.id, cfg.peer_count);
+    node.discover_and_connect();
+    node.wait_for_connections(cfg.peer_count - 1, std::chrono::milliseconds(5000));
+
+    spdlog::info("[p2p] node {}: connections up — starting simulation", cfg.id);
+
+    // ── simulation loop (fixed 60 Hz) ─────────────────────────────────────────
+
+    tw::LockStep lock_step(60);
+
+    while (g_running.load()) {
+        if (lock_step.wait_for_next_step()) continue;
+
+        node.tick();
+
+        if (cfg.ticks > 0 &&
+            static_cast<int>(node.stepped_frames()) >= cfg.ticks)
+        {
+            break;
+        }
+    }
+
+    // ── summary ───────────────────────────────────────────────────────────────
+
+    spdlog::info("[p2p] node {}  frames={} rollbacks={} rollback_rate={:.1f}%",
+                 cfg.id,
+                 node.stepped_frames(),
+                 node.rollbacks(),
+                 100.0 * static_cast<double>(node.rollbacks()) /
+                     static_cast<double>(node.stepped_frames() + 1));
+
+    return 0;
+}

modules/peer_to_peer/src/PeerWorldController.cpp

@@ -0,0 +1,155 @@
+#include "PeerWorldController.hpp"
+
+#include "PeerRegistry.hpp"
+#include "WorldStepProcessor.hpp"
+
+#include <cmath>
+#include <limits>
+#include <spdlog/spdlog.h>
+#include <thread>
+
+namespace tw::p2p {
+
+PeerWorldController::PeerWorldController(uint32_t self_id, uint16_t port,
+                                         std::unique_ptr<PeerLink> link,
+                                         const std::string& registry_path,
+                                         World* world)
+    : m_self_id(self_id)
+    , m_port(port)
+    , m_registry_path(registry_path)
+    , m_link(std::move(link))
+    , m_processor(world, self_id)
+{
+    m_link->on_connected([this](uint32_t peer_id) {
+        m_peer_ids.push_back(peer_id);
+        m_peer_acks[peer_id] = 0;
+        m_processor.add_peer(peer_id);
+        spdlog::info("PeerWorldController[{}]: peer {} connected", m_self_id, peer_id);
+    });
+
+    m_link->on_action([this](uint32_t /*peer_id*/, const PeerAction& action) {
+        m_processor.submit_action(action);
+        // Take max: unreliable transport can deliver packets out of order.
+        auto& ack = m_peer_acks[action.peer_id];
+        ack = std::max(action.ack_frame, ack);
+        spdlog::info("PeerWorldController[{}]: peer {} ack={}", m_self_id, action.peer_id, ack);
+    });
+}
+
+// ── registry & connection ─────────────────────────────────────────────────────
+
+void PeerWorldController::register_self() {
+    PeerRegistry reg(m_registry_path);
+    reg.register_self(m_self_id, "127.0.0.1:" + std::to_string(m_port));
+}
+
+bool PeerWorldController::wait_until_peers_registered(uint32_t expected_count,
+                                                       std::chrono::milliseconds timeout)
+{
+    PeerRegistry reg(m_registry_path);
+    auto deadline = std::chrono::steady_clock::now() + timeout;
+    while (std::chrono::steady_clock::now() < deadline) {
+        if (reg.read_all().size() >= expected_count) return true;
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    }
+    spdlog::warn("PeerWorldController[{}]: timed out waiting for {} peers in registry",
+                 m_self_id, expected_count);
+    return false;
+}
+
+void PeerWorldController::discover_and_connect() {
+    PeerRegistry reg(m_registry_path);
+    for (const auto& info : reg.read_all()) {
+        if (info.id == m_self_id || info.id < m_self_id) continue;
+
+        auto colon = info.address.rfind(':');
+        if (colon == std::string::npos) continue;
+
+        std::string host = info.address.substr(0, colon);
+        int         port = std::stoi(info.address.substr(colon + 1));
+        m_link->connect_to(info.id, tw::net::Address{host, port});
+    }
+}
+
+void PeerWorldController::wait_for_connections(uint32_t expected_count,
+                                               std::chrono::milliseconds timeout)
+{
+    auto deadline = std::chrono::steady_clock::now() + timeout;
+    while (m_peer_ids.size() < expected_count &&
+           std::chrono::steady_clock::now() < deadline)
+    {
+        m_link->poll();
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    if (m_peer_ids.size() < expected_count)
+        spdlog::warn("PeerWorldController[{}]: timed out ({}/{} connections)",
+                     m_self_id, m_peer_ids.size(), expected_count);
+}
+
+// ── tick ──────────────────────────────────────────────────────────────────────
+
+bool PeerWorldController::tick() {
+    m_link->poll();
+
+    if(m_processor.committed_frame() >= std::max((int32_t)m_processor.current_frame() - 50, 0)) {
+        uint32_t frame = m_processor.current_frame();
+        mmo::peer::PeerAction local = build_local_action(frame);
+        m_self_history[frame] = local;
+
+        m_processor.submit_action(PeerAction{
+            .peer_id   = m_self_id,
+            .frame_idx = frame,
+            .input     = glm::vec3(local.input().x(), local.input().y(), local.input().z()),
+        });
+    }
+
+    for (uint32_t peer_id : m_peer_ids)
+        send_history_to(peer_id);
+
+    m_link->poll(); // flush outbound, pick up immediate responses
+
+    // Prune history that all peers have acked.
+    if (!m_peer_ids.empty()) {
+        uint32_t min_ack = std::numeric_limits<uint32_t>::max();
+        for (auto& [pid, acked] : m_peer_acks)
+            min_ack = std::min(min_ack, acked);
+        while (!m_self_history.empty() && m_self_history.begin()->first < min_ack)
+            m_self_history.erase(m_self_history.begin());
+    }
+
+    bool has_advanced = m_processor.advance();
+
+    return has_advanced;
+}
+
+// ── helpers ───────────────────────────────────────────────────────────────────
+
+mmo::peer::PeerAction PeerWorldController::build_local_action(uint32_t frame) const {
+    float t = static_cast<float>(frame) *
+              static_cast<float>(WorldStepProcessor::FIXED_DELTA_S);
+
+    mmo::peer::PeerAction a;
+    a.set_frame_idx(frame);
+    a.mutable_input()->set_x(std::cos(t));
+    a.mutable_input()->set_y(0.0f);
+    a.mutable_input()->set_z(std::sin(t));
+    return a;
+}
+
+void PeerWorldController::send_history_to(uint32_t peer_id) {
+    uint32_t acked = m_peer_acks.at(peer_id);
+
+    mmo::peer::PeerActionBatch batch;
+    batch.set_peer_id(m_self_id);
+    batch.set_ack_frame(m_processor.committed_frame());
+
+    for (auto& [frame, action] : m_self_history) {
+        if (frame < acked) continue;
+        *batch.add_actions() = action;
+    }
+
+    if (batch.actions_size() > 0)
+        m_link->send_batch(peer_id, batch);
+}
+
+} // namespace tw::p2p

modules/peer_to_peer/src/PeerWorldController.hpp

@@ -0,0 +1,75 @@
+#pragma once
+
+#include "PeerLink.hpp"
+#include "WorldStepProcessor.hpp"
+#include "Peer.pb.h"
+#include "world/World.hpp"
+
+#include <chrono>
+#include <map>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+namespace tw::p2p {
+
+// Encapsulates the peer-to-peer networking layer and lock-step simulation.
+//
+// Owns the PeerLink (transport), WorldStepProcessor (simulation), peer
+// registry, action history, and ack tracking.  The caller (TestPeerNode)
+// only needs to supply a World and forward a few lifecycle calls.
+class PeerWorldController {
+public:
+    PeerWorldController(uint32_t self_id, uint16_t port,
+                        std::unique_ptr<PeerLink> link,
+                        const std::string& registry_path,
+                        World* world);
+
+    // Write this node's entry into the shared registry.
+    void register_self();
+
+    // Block until at least expected_count entries appear in the registry.
+    bool wait_until_peers_registered(
+        uint32_t expected_count,
+        std::chrono::milliseconds timeout = std::chrono::milliseconds(30000));
+
+    // Initiate outgoing connections to all higher-ID peers in the registry.
+    void discover_and_connect();
+
+    // Accept expected_count incoming connections via the PeerLink listener.
+    void wait_for_connections(
+        uint32_t expected_count,
+        std::chrono::milliseconds timeout = std::chrono::milliseconds(2000));
+
+    // One simulation tick.  Returns true if the world stepped forward.
+    bool tick();
+
+    uint32_t  self_id()        const { return m_self_id; }
+    uint32_t  current_frame()  const { return m_processor.current_frame(); }
+    uint64_t  rollbacks()      const { return m_processor.rollbacks(); }
+    uint64_t  stepped_frames() const { return m_processor.stepped_frames(); }
+
+    glm::vec3 peer_position(uint32_t peer_id) const {
+        return m_processor.peer_position(peer_id);
+    }
+
+    const std::vector<uint32_t>& peer_ids() const { return m_peer_ids; }
+
+private:
+    uint32_t    m_self_id;
+    uint16_t    m_port;
+    std::string m_registry_path;
+
+    std::unique_ptr<PeerLink> m_link;
+    WorldStepProcessor        m_processor;
+
+    std::vector<uint32_t>                     m_peer_ids;
+    std::map<uint32_t, mmo::peer::PeerAction> m_self_history;
+    std::unordered_map<uint32_t, uint32_t>    m_peer_acks;
+
+    mmo::peer::PeerAction build_local_action(uint32_t frame) const;
+    void send_history_to(uint32_t peer_id);
+};
+
+} // namespace tw::p2p

modules/peer_to_peer/src/QuicrPeerLink.cpp

@@ -0,0 +1,115 @@
+#include "QuicrPeerLink.hpp"
+#include "protocol/quicr/QuicrConnectionListener.hpp"
+#include "protocol/quicr/QuicrEndpoint.hpp"
+
+#include <cstring>
+#include <spdlog/spdlog.h>
+
+namespace tw::p2p {
+using net::quicr::QuicrEndpoint;
+
+QuicrPeerLink::QuicrPeerLink(uint32_t self_id, uint16_t port)
+    : m_self_id(self_id),
+      m_endpoint(QuicrEndpoint::create_and_bind(port).value()),
+      m_listener(net::quicr::QuicrConnectionListener::listen(m_endpoint.get()).value())
+{}
+
+void QuicrPeerLink::connect_to(uint32_t peer_id, const tw::net::Address& addr) {
+    auto r = m_endpoint->connect(addr);
+    if (!r) {
+        spdlog::warn("QuicrPeerLink[{}]: connect to peer {} failed", m_self_id, peer_id);
+        return;
+    }
+    auto* conn = *r;
+    mmo::peer::PeerHello hello;
+    hello.set_peer_id(m_self_id);
+    send_raw(conn, MsgType::Hello, hello, /*reliable=*/true);
+    m_conns.push_back(Conn{conn, peer_id});
+}
+
+void QuicrPeerLink::send_batch(uint32_t peer_id, const mmo::peer::PeerActionBatch& batch) {
+    auto it = m_by_id.find(peer_id);
+    if (it == m_by_id.end()) return;
+    send_raw(it->second, MsgType::ActionBatch, batch, /*reliable=*/false);
+}
+
+void QuicrPeerLink::poll() {
+    m_endpoint->poll();
+    drain_listener();
+    poll_conns();
+}
+
+void QuicrPeerLink::drain_listener() {
+    while (auto* conn = m_listener->listen()) {
+        spdlog::error("PRDIDKI");
+        mmo::peer::PeerHello hello;
+        hello.set_peer_id(m_self_id);
+        send_raw(conn, MsgType::Hello, hello, /*reliable=*/true);
+        m_conns.push_back(Conn{conn, std::nullopt});
+    }
+}
+
+void QuicrPeerLink::poll_conns() {
+    for (auto& c : m_conns) {
+        auto r = c.raw->read_into(std::span(c.recv_buf));
+        if (r && *r > 0)
+            dispatch(c, std::span<const std::byte>(c.recv_buf.data(), *r));
+    }
+}
+
+void QuicrPeerLink::dispatch(Conn& c, std::span<const std::byte> frame) {
+    if (frame.size() < sizeof(uint32_t)) return;
+
+    uint32_t tag;
+    std::memcpy(&tag, frame.data(), sizeof(tag));
+    auto payload = frame.subspan(sizeof(tag));
+
+    switch (static_cast<MsgType>(tag)) {
+        case MsgType::Hello: {
+            mmo::peer::PeerHello msg;
+            if (!msg.ParseFromArray(payload.data(), static_cast<int>(payload.size()))) break;
+            uint32_t remote_id = msg.peer_id();
+            c.peer_id = remote_id;
+            m_by_id[remote_id] = c.raw;
+            if (m_connected_handler) m_connected_handler(remote_id);
+            break;
+        }
+        case MsgType::ActionBatch: {
+            if (!c.peer_id) break;
+            mmo::peer::PeerActionBatch batch;
+            if (!batch.ParseFromArray(payload.data(), static_cast<int>(payload.size()))) break;
+            if (m_action_handler) {
+                for (const auto& pa : batch.actions()) {
+                    m_action_handler(*c.peer_id, PeerAction{
+                        .peer_id   = batch.peer_id(),
+                        .frame_idx = pa.frame_idx(),
+                        .input     = {pa.input().x(), pa.input().y(), pa.input().z()},
+                        .ack_frame = batch.ack_frame(),
+                    });
+                }
+            }
+            break;
+        }
+        case MsgType::Bye:
+            spdlog::debug("QuicrPeerLink[{}]: bye from peer {}", m_self_id, c.peer_id.value_or(0));
+            break;
+        default:
+            spdlog::warn("QuicrPeerLink[{}]: unknown msg type {}", m_self_id, tag);
+    }
+}
+
+void QuicrPeerLink::send_raw(tw::net::quicr::QuicrConnection* conn,
+                              MsgType type,
+                              const google::protobuf::MessageLite& msg,
+                              bool reliable)
+{
+    std::string payload = msg.SerializeAsString();
+    std::vector<std::byte> frame(sizeof(uint32_t) + payload.size());
+    uint32_t tag = static_cast<uint32_t>(type);
+    std::memcpy(frame.data(), &tag, sizeof(tag));
+    std::memcpy(frame.data() + sizeof(tag), payload.data(), payload.size());
+    if (!conn->send_message(std::span(frame), reliable))
+        spdlog::warn("QuicrPeerLink[{}]: send failed", m_self_id);
+}
+
+} // namespace tw::p2p

modules/peer_to_peer/src/QuicrPeerLink.hpp

@@ -0,0 +1,46 @@
+#pragma once
+
+#include "PeerLink.hpp"
+#include "protocol/quicr/QuicrConnection.hpp"
+#include "protocol/quicr/QuicrConnectionListener.hpp"
+#include "protocol/quicr/QuicrEndpoint.hpp"
+
+#include <list>
+#include <optional>
+#include <unordered_map>
+#include <vector>
+
+namespace tw::p2p {
+
+class QuicrPeerLink : public PeerLink {
+public:
+    QuicrPeerLink(uint32_t self_id, uint16_t port);
+
+    void connect_to(uint32_t peer_id, const tw::net::Address& addr) override;
+    void send_batch(uint32_t peer_id, const mmo::peer::PeerActionBatch& batch) override;
+    void poll      () override;
+
+private:
+    enum class MsgType : uint32_t { Hello = 1, ActionBatch = 2, Bye = 3 };
+
+    struct Conn {
+        tw::net::quicr::QuicrConnection* raw;
+        std::optional<uint32_t>          peer_id;
+        std::vector<std::byte>           recv_buf{64 * 1024};
+    };
+
+    uint32_t m_self_id;
+    std::unique_ptr<tw::net::quicr::QuicrEndpoint>           m_endpoint;
+    std::unique_ptr<tw::net::quicr::QuicrConnectionListener> m_listener;
+
+    std::list<Conn> m_conns;
+    std::unordered_map<uint32_t, tw::net::quicr::QuicrConnection*> m_by_id;
+
+    void send_raw     (tw::net::quicr::QuicrConnection* conn, MsgType type,
+                       const google::protobuf::MessageLite& msg, bool reliable);
+    void dispatch     (Conn& c, std::span<const std::byte> frame);
+    void drain_listener();
+    void poll_conns   ();
+};
+
+} // namespace tw::p2p

modules/peer_to_peer/src/TcpPeerLink.cpp

@@ -0,0 +1,143 @@
+#include "TcpPeerLink.hpp"
+
+#include <cstring>
+#include <spdlog/spdlog.h>
+
+namespace tw::p2p {
+
+TcpPeerLink::TcpPeerLink(uint32_t self_id, const tw::net::Address& listen_addr)
+    : m_self_id(self_id)
+    , m_listener(tw::net::TcpListener::listen(listen_addr, listen_addr.port()).value())
+{
+    (void)m_listener.set_non_blocking();
+}
+
+void TcpPeerLink::connect_to(uint32_t peer_id, const tw::net::Address& addr) {
+    auto r = tw::net::TcpStream::connect(addr);
+    if (!r) {
+        spdlog::warn("TcpPeerLink[{}]: connect to peer {} failed", m_self_id, peer_id);
+        return;
+    }
+    (void)r->set_non_blocking();
+
+    mmo::peer::PeerHello hello;
+    hello.set_peer_id(m_self_id);
+    send_raw(*r, MsgType::Hello, hello);
+
+    m_conns.push_back(Conn{std::move(*r), peer_id});
+}
+
+void TcpPeerLink::send_batch(uint32_t peer_id, const mmo::peer::PeerActionBatch& batch) {
+    auto it = m_by_id.find(peer_id);
+    if (it == m_by_id.end()) return;
+    send_raw(*it->second, MsgType::ActionBatch, batch);
+}
+
+void TcpPeerLink::poll() {
+    drain_listener();
+    poll_conns();
+}
+
+void TcpPeerLink::drain_listener() {
+    while (true) {
+        auto r = m_listener.listen();
+        if (!r) break;
+        (void)r->set_non_blocking();
+
+        mmo::peer::PeerHello hello;
+        hello.set_peer_id(m_self_id);
+        send_raw(*r, MsgType::Hello, hello);
+
+        m_conns.push_back(Conn{std::move(*r), std::nullopt});
+    }
+}
+
+void TcpPeerLink::poll_conns() {
+    for (auto& c : m_conns) {
+        auto space = std::span(c.recv_buf.data() + c.recv_filled,
+                               c.recv_buf.size() - c.recv_filled);
+        auto r = c.stream.read_into(space);
+        if (r && *r > 0) {
+            c.recv_filled += *r;
+            dispatch(c);
+        }
+    }
+}
+
+void TcpPeerLink::dispatch(Conn& c) {
+    size_t pos = 0;
+    while (c.recv_filled - pos >= sizeof(uint32_t)) {
+        uint32_t total_len;
+        std::memcpy(&total_len, c.recv_buf.data() + pos, sizeof(total_len));
+
+        if (c.recv_filled - pos < sizeof(uint32_t) + total_len) break;
+
+        pos += sizeof(uint32_t);
+        auto msg_span = std::span(c.recv_buf.data() + pos, total_len);
+
+        if (total_len >= sizeof(uint32_t)) {
+            uint32_t tag;
+            std::memcpy(&tag, msg_span.data(), sizeof(tag));
+            auto payload = msg_span.subspan(sizeof(tag));
+
+            switch (static_cast<MsgType>(tag)) {
+                case MsgType::Hello: {
+                    mmo::peer::PeerHello msg;
+                    if (!msg.ParseFromArray(payload.data(), static_cast<int>(payload.size()))) break;
+                    uint32_t remote_id = msg.peer_id();
+                    c.peer_id = remote_id;
+                    m_by_id[remote_id] = &c.stream;
+                    if (m_connected_handler) m_connected_handler(remote_id);
+                    break;
+                }
+                case MsgType::ActionBatch: {
+                    if (!c.peer_id) break;
+                    mmo::peer::PeerActionBatch batch;
+                    if (!batch.ParseFromArray(payload.data(), static_cast<int>(payload.size()))) break;
+                    if (m_action_handler) {
+                        for (const auto& pa : batch.actions()) {
+                            m_action_handler(*c.peer_id, PeerAction{
+                                .peer_id   = batch.peer_id(),
+                                .frame_idx = pa.frame_idx(),
+                                .input     = {pa.input().x(), pa.input().y(), pa.input().z()},
+                                .ack_frame = batch.ack_frame(),
+                            });
+                        }
+                    }
+                    break;
+                }
+                case MsgType::Bye:
+                    spdlog::debug("TcpPeerLink[{}]: bye from peer {}",
+                                  m_self_id, c.peer_id.value_or(0));
+                    break;
+                default:
+                    spdlog::warn("TcpPeerLink[{}]: unknown msg type {}", m_self_id, tag);
+            }
+        }
+        pos += total_len;
+    }
+
+    if (pos > 0) {
+        std::memmove(c.recv_buf.data(), c.recv_buf.data() + pos, c.recv_filled - pos);
+        c.recv_filled -= pos;
+    }
+}
+
+void TcpPeerLink::send_raw(tw::net::TcpStream& stream,
+                            MsgType type,
+                            const google::protobuf::MessageLite& msg)
+{
+    std::string payload  = msg.SerializeAsString();
+    uint32_t total_len   = static_cast<uint32_t>(sizeof(uint32_t) + payload.size());
+    uint32_t tag         = static_cast<uint32_t>(type);
+
+    std::vector<std::byte> frame(sizeof(uint32_t) * 2 + payload.size());
+    std::memcpy(frame.data(),                      &total_len, sizeof(total_len));
+    std::memcpy(frame.data() + sizeof(uint32_t),   &tag,       sizeof(tag));
+    std::memcpy(frame.data() + sizeof(uint32_t)*2, payload.data(), payload.size());
+
+    if (!stream.write(std::span(frame)))
+        spdlog::warn("TcpPeerLink[{}]: send failed", m_self_id);
+}
+
+} // namespace tw::p2p

modules/peer_to_peer/src/TcpPeerLink.hpp

@@ -0,0 +1,51 @@
+#pragma once
+
+#include "PeerLink.hpp"
+#include "TcpListener.hpp"
+#include "TcpStream.hpp"
+
+#include <list>
+#include <optional>
+#include <unordered_map>
+#include <vector>
+
+namespace tw::p2p {
+
+// TCP implementation of PeerLink.
+//
+// Wire format (stream transport requires length-prefix framing):
+//   [uint32_t total_len][uint32_t type][protobuf payload]
+//   total_len = sizeof(type) + sizeof(payload)
+class TcpPeerLink : public PeerLink {
+public:
+    TcpPeerLink(uint32_t self_id, const tw::net::Address& listen_addr);
+
+    void connect_to(uint32_t peer_id, const tw::net::Address& addr) override;
+    void send_batch(uint32_t peer_id, const mmo::peer::PeerActionBatch& batch) override;
+    void poll      () override;
+
+private:
+    enum class MsgType : uint32_t { Hello = 1, ActionBatch = 2, Bye = 3 };
+
+    struct Conn {
+        tw::net::TcpStream      stream;
+        std::optional<uint32_t> peer_id;
+        std::vector<std::byte>  recv_buf{64 * 1024};
+        size_t                  recv_filled{0};
+    };
+
+    uint32_t             m_self_id;
+    tw::net::TcpListener m_listener;
+
+    // std::list: nodes are pointer-stable so m_by_id can safely point into it.
+    std::list<Conn> m_conns;
+    std::unordered_map<uint32_t, tw::net::TcpStream*> m_by_id;
+
+    void send_raw    (tw::net::TcpStream& stream, MsgType type,
+                      const google::protobuf::MessageLite& msg);
+    void dispatch    (Conn& c);
+    void drain_listener();
+    void poll_conns  ();
+};
+
+} // namespace tw::p2p

modules/peer_to_peer/src/TestPeerNode.cpp

@@ -0,0 +1,58 @@
+#include "TestPeerNode.hpp"
+
+#include <chrono>
+#include <spdlog/spdlog.h>
+
+namespace tw::p2p {
+
+TestPeerNode::TestPeerNode(uint32_t self_id, uint16_t port,
+                           std::unique_ptr<PeerLink> link,
+                           const std::string& registry_path,
+                           const std::string& csv_path)
+    : m_world()
+    , m_controller(self_id, port, std::move(link), registry_path, &m_world)
+    , m_csv(csv_path)
+{
+    m_csv << "entity_id,x,y,z,timestamp_ms\n";
+}
+
+void TestPeerNode::register_self() {
+    m_controller.register_self();
+}
+
+bool TestPeerNode::wait_until_peers_registered(uint32_t expected_count,
+                                                std::chrono::milliseconds timeout)
+{
+    return m_controller.wait_until_peers_registered(expected_count, timeout);
+}
+
+void TestPeerNode::discover_and_connect() {
+    m_controller.discover_and_connect();
+}
+
+void TestPeerNode::wait_for_connections(uint32_t expected_count,
+                                        std::chrono::milliseconds timeout)
+{
+    m_controller.wait_for_connections(expected_count, timeout);
+}
+
+void TestPeerNode::tick() {
+    if (m_controller.tick()) {}
+    write_csv_row();
+}
+
+void TestPeerNode::write_csv_row() {
+    auto now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+        std::chrono::system_clock::now().time_since_epoch()).count();
+
+    uint32_t self = m_controller.self_id();
+    glm::vec3 sp  = m_controller.peer_position(self);
+    m_csv << self << "," << sp.x << "," << sp.y << "," << sp.z << "," << now_ms << "\n";
+
+    for (uint32_t pid : m_controller.peer_ids()) {
+        glm::vec3 p = m_controller.peer_position(pid);
+        m_csv << pid << "," << p.x << "," << p.y << "," << p.z << "," << now_ms << "\n";
+    }
+}
+
+} // namespace tw::p2p

modules/peer_to_peer/src/TestPeerNode.hpp

@@ -0,0 +1,61 @@
+#pragma once
+
+#include "PeerLink.hpp"
+#include "PeerWorldController.hpp"
+#include "world/World.hpp"
+
+#include <chrono>
+#include <fstream>
+#include <memory>
+#include <string>
+
+namespace tw::p2p {
+
+// Thin shell around World + PeerWorldController.
+//
+// Owns the World (ECS) and the controller that drives networking and
+// simulation.  The only logic here is CSV output.
+class TestPeerNode {
+public:
+    TestPeerNode(uint32_t self_id, uint16_t port,
+                 std::unique_ptr<PeerLink> link,
+                 const std::string& registry_path,
+                 const std::string& csv_path);
+
+    TestPeerNode(const TestPeerNode&) = delete;
+    TestPeerNode& operator=(const TestPeerNode&) = delete;
+
+    ~TestPeerNode() { m_csv.close(); }
+
+    void register_self();
+
+    bool wait_until_peers_registered(
+        uint32_t expected_count,
+        std::chrono::milliseconds timeout = std::chrono::milliseconds(30000));
+
+    void discover_and_connect();
+
+    void wait_for_connections(
+        uint32_t expected_count,
+        std::chrono::milliseconds timeout = std::chrono::milliseconds(2000));
+
+    // Advance one tick; writes a CSV row if the world stepped.
+    void tick();
+
+    uint32_t  self_id()        const { return m_controller.self_id(); }
+    uint32_t  current_frame()  const { return m_controller.current_frame(); }
+    uint64_t  rollbacks()      const { return m_controller.rollbacks(); }
+    uint64_t  stepped_frames() const { return m_controller.stepped_frames(); }
+    glm::vec3 peer_position(uint32_t peer_id) const {
+        return m_controller.peer_position(peer_id);
+    }
+
+private:
+    tw::World           m_world;
+    PeerWorldController m_controller;
+    std::ofstream       m_csv;
+
+    void write_csv_row();
+};
+
+} // namespace tw::p2p

modules/peer_to_peer/src/WorldStepProcessor.cpp

@@ -0,0 +1,123 @@
+#include "WorldStepProcessor.hpp"
+#include "world/Transform.hpp"
+
+namespace tw::p2p {
+
+WorldStepProcessor::WorldStepProcessor(World* world, uint32_t self_id)
+    : m_world(world)
+    , m_self_id(self_id)
+{
+    m_all_peer_ids.insert(self_id);
+    auto entity = m_world->registry().create();
+    m_world->registry().emplace<tw::Transform>(entity, glm::vec3(0.0f));
+    m_peer_entities[self_id] = entity;
+    m_committed_snapshot.positions[self_id] = glm::vec3(0.0f);
+}
+
+void WorldStepProcessor::add_peer(uint32_t peer_id) {
+    if (m_all_peer_ids.count(peer_id)) return;
+    m_all_peer_ids.insert(peer_id);
+    auto entity = m_world->registry().create();
+    m_world->registry().emplace<tw::Transform>(entity, glm::vec3(0.0f));
+    m_peer_entities[peer_id] = entity;
+    m_committed_snapshot.positions[peer_id] = glm::vec3(0.0f);
+}
+
+// ── submit ────────────────────────────────────────────────────────────────────
+
+void WorldStepProcessor::submit_action(const PeerAction& action) {
+    m_frames[action.frame_idx].actions[action.peer_id] = action;
+}
+
+// ── advance ───────────────────────────────────────────────────────────────────
+
+bool WorldStepProcessor::advance() {
+    flush_commits();
+
+    Frame empty;
+    const Frame& frame = m_frames.count(m_current_frame)
+                             ? m_frames.at(m_current_frame)
+                             : empty;
+    apply_frame(frame);
+    ++m_current_frame;
+    ++m_stepped_frames;
+    return true;
+}
+
+// ── flush_commits ─────────────────────────────────────────────────────────────
+
+void WorldStepProcessor::flush_commits() {
+    const uint32_t old_committed = m_committed_frame;
+
+    while (m_frames.count(m_committed_frame) &&
+           m_frames.at(m_committed_frame).is_full(m_all_peer_ids))
+    {
+        ++m_committed_frame;
+    }
+
+    if (m_committed_frame == old_committed) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        return;
+    }
+
+    // Advance the committed snapshot through the newly committed frames.
+    restore_snapshot(m_committed_snapshot);
+    for (uint32_t f = old_committed; f < m_committed_frame; ++f)
+        apply_frame(m_frames.at(f));
+    m_committed_snapshot = capture_snapshot();
+
+    // Re-simulate any speculative frames on top of the new committed state.
+    for (uint32_t f = m_committed_frame; f < m_current_frame; ++f) {
+        Frame empty;
+        const Frame& fr = m_frames.count(f) ? m_frames.at(f) : empty;
+        apply_frame(fr);
+    }
+
+    // Prune frames that are now part of the committed snapshot.
+    while (!m_frames.empty() && m_frames.begin()->first < m_committed_frame)
+        m_frames.erase(m_frames.begin());
+
+    ++m_rollbacks;
+}
+
+// ── helpers ───────────────────────────────────────────────────────────────────
+
+void WorldStepProcessor::apply_frame(const Frame& frame) {
+    for (uint32_t peer_id : m_all_peer_ids) {
+        glm::vec3 input(0.0f);
+        auto it = frame.actions.find(peer_id);
+        if (it != frame.actions.end())
+            input = it->second.input;
+
+        auto* t = m_world->registry().try_get<tw::Transform>(m_peer_entities.at(peer_id));
+        if (t) t->translate(input * MOVE_SPEED * static_cast<float>(FIXED_DELTA_S));
+    }
+    m_world->step(FIXED_DELTA_S);
+}
+
+WorldStepProcessor::Snapshot WorldStepProcessor::capture_snapshot() const {
+    Snapshot snap;
+    for (auto& [peer_id, entity] : m_peer_entities) {
+        const auto* t = m_world->registry().try_get<tw::Transform>(entity);
+        snap.positions[peer_id] = t ? t->position() : glm::vec3(0.0f);
+    }
+    return snap;
+}
+
+void WorldStepProcessor::restore_snapshot(const Snapshot& snap) {
+    for (auto& [peer_id, pos] : snap.positions) {
+        auto it = m_peer_entities.find(peer_id);
+        if (it == m_peer_entities.end()) continue;
+        auto* t = m_world->registry().try_get<tw::Transform>(it->second);
+        if (t) t->set_position(pos);
+    }
+}
+
+glm::vec3 WorldStepProcessor::peer_position(uint32_t peer_id) const {
+    auto it = m_peer_entities.find(peer_id);
+    if (it == m_peer_entities.end()) return glm::vec3(0.0f);
+    const auto* t = m_world->registry().try_get<tw::Transform>(it->second);
+    return t ? t->position() : glm::vec3(0.0f);
+}
+
+} // namespace tw::p2p

modules/peer_to_peer/src/WorldStepProcessor.hpp

@@ -0,0 +1,92 @@
+#pragma once
+
+#include "PeerAction.hpp"
+#include "world/World.hpp"
+
+#include <entt/entt.hpp>
+#include <map>
+#include <unordered_map>
+#include <unordered_set>
+
+namespace tw::p2p {
+
+// Rollback-based lock-step processor.
+//
+// Speculative advance
+//   try_advance() always steps the world.  For each peer whose action has not
+//   yet arrived the entity is held stationary (zero input — no prediction).
+//
+// Committed frames
+//   A Frame becomes committed once every registered peer (including self) has
+//   submitted an action for it.  try_advance() cascade-commits all newly-full
+//   frames in one pass before stepping.  If any new frames were committed the
+//   world is rolled back to the last committed snapshot and the remaining
+//   speculative frames are re-simulated with whatever confirmed actions exist.
+//
+// One re-simulation per tick (not per submit_action).
+class WorldStepProcessor {
+public:
+    static constexpr double FIXED_DELTA_S = (1000.0 / 60.0) / 1000.0;
+    static constexpr float  MOVE_SPEED    = 5.0f;
+
+    WorldStepProcessor(World* world, uint32_t self_id);
+
+    void add_peer(uint32_t peer_id);
+
+    // Store a confirmed action.  Does not touch the world or trigger rollback.
+    void submit_action(const PeerAction& action);
+
+    // Cascade-commit full frames, rollback + re-simulate if needed, then
+    // advance one speculative frame.  Always returns true.
+    bool advance();
+
+    uint32_t current_frame()   const { return m_current_frame; }
+    uint32_t committed_frame() const { return m_committed_frame; }
+    uint64_t rollbacks()       const { return m_rollbacks; }
+    uint64_t stepped_frames()  const { return m_stepped_frames; }
+
+    glm::vec3 peer_position(uint32_t peer_id) const;
+
+private:
+    // All confirmed actions for one simulation frame.
+    struct Frame {
+        std::unordered_map<uint32_t, PeerAction> actions; // peer_id -> action
+
+        bool is_full(const std::unordered_set<uint32_t>& peers) const {
+            for (uint32_t id : peers)
+                if (!actions.count(id)) return false;
+            return true;
+        }
+    };
+
+    // World-state snapshot: positions of every tracked entity.
+    struct Snapshot {
+        std::unordered_map<uint32_t, glm::vec3> positions;
+    };
+
+    World*   m_world;
+    uint32_t m_self_id;
+
+    uint32_t m_current_frame   = 0; // next speculative frame to advance
+    uint32_t m_committed_frame = 0; // next frame eligible to commit
+
+    Snapshot m_committed_snapshot; // world state before m_committed_frame
+
+    std::unordered_set<uint32_t>               m_all_peer_ids;
+    std::unordered_map<uint32_t, entt::entity> m_peer_entities;
+    std::map<uint32_t, Frame>                  m_frames; // partial or full
+
+    uint64_t m_rollbacks      = 0;
+    uint64_t m_stepped_frames = 0;
+
+    // Apply all actions in frame to the current world state; zero for missing.
+    void apply_frame(const Frame& frame);
+
+    // Cascade-commit full frames and re-simulate speculative ones if needed.
+    void flush_commits();
+
+    Snapshot capture_snapshot() const;
+    void     restore_snapshot (const Snapshot& snap);
+};
+
+} // namespace tw::p2p