voxelforge/core/include/vf/Transport.h

#pragma once

#include <cstddef>
#include <cstdint>
#include <deque>
#include <map>
#include <vector>
#include "Codec.h"

namespace vf {

// Datagram numbering is 16-bit and wraps, so every comparison between two
// sequence numbers has to be modular rather than arithmetic.
using SequenceNumber = uint16_t;

// Bit i means "the datagram numbered ackLatest - 1 - i also arrived", so one
// 32-bit word confirms 33 datagrams and a single surviving ack can recover the
// whole window that came before it.
using AckBitfield = uint32_t;

// Everything a hostile link does to a datagram, as integer percents and tick
// counts. The seed makes a run reproducible: the same sequence of sends always
// produces the same sequence of arrivals.
struct LinkQuality {
    uint32_t lossPercent = 0;
    uint32_t duplicatePercent = 0;
    uint32_t reorderPercent = 0;
    uint32_t latencyTicks = 1;
    uint32_t reorderExtraTicks = 3;
    uint64_t seed = 0x9e3779b97f4a7c15ull;
};

// A two-endpoint in-memory pipe: no sockets, no threads, no wall clock.
// Delivery is a tick-stamped queue and every impairment is drawn from a seeded
// stream, which is what makes the transport tests reproducible.
class LoopbackLink {
public:
    explicit LoopbackLink(LinkQuality quality = {});

    void send(int fromEndpoint, const uint8_t* datagram, std::size_t size, uint64_t tick);
    void send(int fromEndpoint, const ByteBuffer& datagram, uint64_t tick);

    // Every datagram addressed to `endpoint` whose delivery tick has arrived,
    // oldest first. Same-tick arrivals keep their send order.
    void receive(int endpoint, uint64_t tick, std::vector<ByteBuffer>& out);

    // Re-seeds the impairment stream, so a test can change the link mid-run.
    void setQuality(const LinkQuality& newQuality);
    const LinkQuality& settings() const { return quality; }
    void reset();

    uint64_t sentCount() const { return sent; }
    uint64_t droppedCount() const { return dropped; }
    uint64_t duplicatedCount() const { return duplicated; }
    uint64_t delayedCount() const { return delayed; }
    std::size_t inFlightCount() const { return pending.size(); }

private:
    struct PendingDatagram {
        uint64_t deliveryTick = 0;
        uint64_t order = 0;
        int toEndpoint = 0;
        ByteBuffer bytes;
    };

    uint64_t nextRandom();
    bool roll(uint32_t percent);
    void enqueue(int toEndpoint, const ByteBuffer& bytes, uint64_t deliveryTick);

    LinkQuality quality;
    uint64_t rngState;
    uint64_t orderCounter = 0;
    std::vector<PendingDatagram> pending;
    uint64_t sent = 0;
    uint64_t dropped = 0;
    uint64_t duplicated = 0;
    uint64_t delayed = 0;
};

// One unit on the wire: an ack header plus however many message fragments fit
// under the MTU. A fragment carries its own message identity so a datagram can
// be lost without losing track of what was in it.
struct Datagram {
    struct Fragment {
        bool reliable = false;
        uint32_t messageId = 0;
        uint32_t fragmentIndex = 0;
        uint32_t fragmentCount = 1;
        ByteBuffer bytes;
    };

    SequenceNumber sequence = 0;
    SequenceNumber ackLatest = 0;
    AckBitfield ackHistory = 0;
    // Until the sender has itself received something, ackLatest is meaningless
    // and must not be mistaken for an acknowledgement of sequence 0.
    bool ackValid = false;
    std::vector<Fragment> fragments;

    void encode(ByteWriter& out) const;
    bool decode(const uint8_t* data, std::size_t size);

    static std::size_t headerSize() { return 11; }
    static std::size_t maxFragmentOverhead() { return 21; }
};

struct TransportConfig {
    std::size_t mtu = 512;
    uint32_t retransmitTimeoutTicks = 8;
    uint32_t sendWindow = 32;
    uint32_t rttSmoothingPercent = 20;
};

struct TransportStats {
    uint64_t datagramsSent = 0;
    uint64_t datagramsReceived = 0;
    uint64_t datagramsLost = 0;
    uint64_t retransmitted = 0;
    uint64_t duplicatesDropped = 0;
    uint64_t staleDropped = 0;
    uint64_t messagesQueued = 0;
    uint64_t messagesDelivered = 0;
    uint64_t bytesSent = 0;
    uint64_t bytesReceived = 0;
    double smoothedRttTicks = 0.0;
};

// Ordered, exactly-once message stream. Outgoing fragments live until acked;
// incoming fragments are reassembled per message and then held back until
// every earlier message id has been handed over.
class ReliableChannel {
public:
    struct OutgoingFragment {
        uint32_t messageId = 0;
        uint32_t fragmentIndex = 0;
        uint32_t fragmentCount = 1;
        ByteBuffer bytes;
        bool inFlight = false;
    };

    void queueMessage(const uint8_t* payload, std::size_t size, std::size_t fragmentPayload);

    void collectReady(std::vector<uint64_t>& keys) const;
    const OutgoingFragment* fragment(uint64_t key) const;
    void markInFlight(uint64_t key);
    void markAcked(uint64_t key);
    void markLost(uint64_t key);

    std::size_t unackedFragmentCount() const { return outgoing.size(); }
    std::size_t readyFragmentCount() const;

    // False when the fragment belongs to a message already handed over, which
    // is what a retransmission of an already-delivered message looks like.
    bool accept(const Datagram::Fragment& fragment);
    bool popDelivered(ByteBuffer& out);
    std::size_t deliveredWaiting() const { return delivered.size(); }
    uint32_t nextExpectedMessageId() const { return nextDeliverMessageId; }
    std::size_t reorderBufferSize() const { return completed.size(); }