// The serve pump: single-flight per peer, and parking on a full write buffer
// instead of producing into it.
//
// WHY THIS MATTERS BEYOND THIS PROTOCOL. Sibling protocols share the same Noise
// stream — the game's own RPC channel, DHT traffic — and Protomux's backlog is
// unbounded. A chunk pump that ignores writability does not fail; it succeeds,
// and takes the rest of the connection down with it. Nothing about that is
// visible in the frames, so it gets asserted here or nowhere.
//
// Build:
// g++ -O2 -std=c++17 -I liboverlay/include $(pkg-config --cflags libsodium) \
// -o /tmp/ho_bp liboverlay/test/test_backpressure.cpp $(pkg-config --libs libsodium)
#include <cstdio>
#include <set>
#include <string>
#include <vector>
#include "overlay/protocol.h"
using namespace hyperoverlay;
static int passed = 0, failed = 0;
static void check(bool c, const std::string &what) {
if (c) { passed++; printf(" ok %s\n", what.c_str()); }
else { failed++; printf(" FAIL %s\n", what.c_str()); }
}
static const char *TMP = "/tmp/claude-1000/";
static std::string tmp_path(const std::string &n) { return std::string(TMP) + n; }
static std::vector<uint8_t> make_bytes(size_t n, uint32_t seed) {
std::vector<uint8_t> v(n);
uint32_t s = seed;
for (size_t i = 0; i < n; i++) { s = s * 1664525u + 1013904223u; v[i] = (uint8_t)(s >> 24); }
return v;
}
// A carrier whose buffer fills after `capacity` frames and stays full until the
// test drains it — the shape libudx reports at its high watermark.
struct Wire : Transport {
std::string name;
std::vector<std::pair<MsgId, std::vector<uint8_t>>> sent;
int capacity = -1; // <0 = infinite
int since_drain = 0;
// A carrier that delivers INLINE: send() re-enters the protocol with the
// peer's next request, the way an in-memory pair does (and the way the JS
// reference harness pairs two overlays). One-shot, armed by the test.
Protocol *reenter_into = nullptr;
std::vector<uint8_t> reenter_frame;
bool reenter_armed = false;
explicit Wire(const char *n) : name(n) {}
bool send(MsgId id, const std::vector<uint8_t> &f) override {
sent.push_back({ id, f });
if (reenter_armed && id == MsgId::ChunkData) {
reenter_armed = false;
reenter_into->on_message(this, MsgId::ChunkNeed, reenter_frame);
}
if (capacity < 0) return true;
return ++since_drain < capacity;
}
const char *peer_label() const override { return name.c_str(); }
void drain() { since_drain = 0; }
int count(MsgId want) const {
int n = 0;
for (const auto &e : sent) if (e.first == want) n++;
return n;
}
std::vector<uint64_t> data_indices() const {
std::vector<uint64_t> out;
for (const auto &e : sent) {
if (e.first != MsgId::ChunkData) continue;
cenc::State s;
s.buffer = const_cast<uint8_t *>(e.second.data());
s.start = 0; s.end = e.second.size();
out.push_back(msg::decode_chunk_data(s).index);
}
return out;
}
};
int main() {
printf("hyper-overlay C++ serve pump (backpressure)\n");
const std::vector<uint8_t> content = make_bytes(300 * 1024, 91);
uint8_t oid_raw[32];
cdc::hash_chunk(content.data(), content.size(), oid_raw);
const std::string oid = to_hex(oid_raw, 32);
const std::string src = tmp_path("ho_bp_src.bin");
{
FILE *f = std::fopen(src.c_str(), "wb");
std::fwrite(content.data(), 1, content.size(), f);
std::fclose(f);
}
const size_t total_chunks = cdc::chunk(content.data(), content.size()).size();
auto holder = [&](index::FileIndex *ix) {
return new Protocol(ix, [&](const std::string &h, ContentLocation &out) {
if (h != oid) return false;
out.path = src;
return true;
});
};
auto ask_for_list = [&](Protocol &p, Wire &w) {
msg::ContentRequest req;
req.contentHash = from_hex32(oid);
req.from = "tester";
p.on_message(&w, MsgId::ContentRequest, msg::encode(req));
};
auto ask_for_chunks = [&](Protocol &p, Wire &w, const std::vector<uint64_t> &idx) {
msg::ChunkNeed need;
need.path = content_path(oid);
need.indices = idx;
p.on_message(&w, MsgId::ChunkNeed, msg::encode(need));
};
std::vector<uint64_t> all;
for (uint64_t i = 0; i < total_chunks; i++) all.push_back(i);
// ---- the pump parks, and resumes where it stopped -----------------------
{
printf("\n-- parking\n");
index::FileIndex ix;
std::remove(tmp_path("ho_bp1.log").c_str());
ix.open(tmp_path("ho_bp1.log"));
Protocol *alice = holder(&ix);
Wire w("w");
alice->add_peer(&w);
ask_for_list(*alice, w);
check(w.count(MsgId::ChunkHashes) == 1, "the holder answers with a chunk list");
check(total_chunks > 6, "the file is several chunks (" + std::to_string(total_chunks) + ")");
w.sent.clear();
w.capacity = 3;
w.since_drain = 0;
ask_for_chunks(*alice, w, all);
check(w.count(MsgId::ChunkData) == 3,
"the pump stops at the buffer's capacity rather than producing into it (" +
std::to_string(w.count(MsgId::ChunkData)) + " of " + std::to_string(total_chunks) + ")");
// Nothing more until the carrier says so — not on a timer, not on the next
// message.
ask_for_list(*alice, w);
check(w.count(MsgId::ChunkData) == 3, "and stays parked while other traffic arrives");
// Drain repeatedly until it finishes.
int rounds = 0;
while (w.count(MsgId::ChunkData) < (int)total_chunks && rounds++ < 100) {
w.drain();
alice->peer_writable(&w);
}
const std::vector<uint64_t> got = w.data_indices();
check(got.size() == total_chunks, "draining resumes it to completion (" +
std::to_string(got.size()) + "/" + std::to_string(total_chunks) + ")");
// Resuming must CONTINUE, not restart: a pump that re-sends from the top of
// the job doubles its traffic every time the buffer fills.
std::set<uint64_t> unique(got.begin(), got.end());
check(unique.size() == got.size(), "no chunk is sent twice — the job resumes where it parked");
bool ordered = true;
for (size_t i = 0; i < got.size(); i++) if (got[i] != (uint64_t)i) ordered = false;
check(ordered, "and the requested order is preserved across the parks");
delete alice;
}
// ---- single-flight: a second request queues, it does not race -----------
//
// Protomux dispatches every frame of one socket read synchronously, so two
// chunkNeed frames arrive back to back with no chance to breathe between them.
// Two pumps would interleave their sends and honour the buffer once between
// them instead of once each.
{
printf("\n-- single flight\n");
index::FileIndex ix;
std::remove(tmp_path("ho_bp2.log").c_str());
ix.open(tmp_path("ho_bp2.log"));
Protocol *alice = holder(&ix);
Wire w("w");
alice->add_peer(&w);
ask_for_list(*alice, w);
w.sent.clear();
w.capacity = 2;
w.since_drain = 0;
// First job: chunks 0,1,2. Second job, arriving while the first is parked:
// chunks 3,4.
ask_for_chunks(*alice, w, { 0, 1, 2 });
check(w.count(MsgId::ChunkData) == 2, "the first job parks at capacity");
ask_for_chunks(*alice, w, { 3, 4 });
check(w.count(MsgId::ChunkData) == 2,
"a second request arriving while parked sends nothing — it queues");
int rounds = 0;
while (w.count(MsgId::ChunkData) < 5 && rounds++ < 50) {
w.drain();
alice->peer_writable(&w);
}
const std::vector<uint64_t> got = w.data_indices();
check(got.size() == 5, "both jobs complete once the carrier drains");
check(got == std::vector<uint64_t>({ 0, 1, 2, 3, 4 }),
"and they are served in order, not interleaved");
delete alice;
}
// ---- one blocked peer does not block another ----------------------------
{
printf("\n-- per-peer isolation\n");
index::FileIndex ix;
std::remove(tmp_path("ho_bp3.log").c_str());
ix.open(tmp_path("ho_bp3.log"));
Protocol *alice = holder(&ix);
Wire slow("slow"), fast("fast");
alice->add_peer(&slow);
alice->add_peer(&fast);
ask_for_list(*alice, slow);
ask_for_list(*alice, fast);
slow.sent.clear(); fast.sent.clear();
slow.capacity = 1; slow.since_drain = 0; // wedges immediately
fast.capacity = -1; // always writable
ask_for_chunks(*alice, slow, all);
check(slow.count(MsgId::ChunkData) == 1, "the wedged peer parks after one chunk");
ask_for_chunks(*alice, fast, all);
check(fast.count(MsgId::ChunkData) == (int)total_chunks,
"a peer with room is served in full regardless (" +
std::to_string(fast.count(MsgId::ChunkData)) + ")");
check(slow.count(MsgId::ChunkData) == 1, "and the wedged one is still parked, not starved out");
delete alice;
}
// ---- a carrier that delivers inline must not start a second pump --------
//
// This is what the single-flight guard is FOR. A carrier whose send() reaches
// the peer synchronously — an in-memory pair, or any future inline adapter —
// can deliver that peer's next chunkNeed back to us from inside our own send
// call. Without the guard the re-entrant call starts a second pump that
// interleaves its frames with the one already running, and the two honour the
// write buffer once between them instead of once each.
//
// (A parked peer is stopped by the blocked check instead, so this scenario is
// the only one that reaches the guard at all.)
{
printf("\n-- a re-entrant carrier\n");
index::FileIndex ix;
std::remove(tmp_path("ho_bp5.log").c_str());
ix.open(tmp_path("ho_bp5.log"));
Protocol *alice = holder(&ix);
Wire w("w");
alice->add_peer(&w);
ask_for_list(*alice, w);
w.sent.clear();
// While serving 0,1,2 the peer asks for 5,6 from inside our send().
msg::ChunkNeed second;
second.path = content_path(oid);
second.indices = { 5, 6 };
w.reenter_into = alice;
w.reenter_frame = msg::encode(second);
w.reenter_armed = true;
ask_for_chunks(*alice, w, { 0, 1, 2 });
const std::vector<uint64_t> got = w.data_indices();
check(got == std::vector<uint64_t>({ 0, 1, 2, 5, 6 }),
"the re-entrant request is queued and served AFTER the running job, not spliced into it");
std::set<uint64_t> uniq(got.begin(), got.end());
check(uniq.size() == got.size(), "and nothing is sent twice");
delete alice;
}
// ---- a cancelled chunk is skipped, not served ---------------------------
//
// The downloader hotswapped it to a faster holder. Serving it anyway spends
// bandwidth on bytes it is already receiving elsewhere.
{
printf("\n-- hotswap cancel\n");
index::FileIndex ix;
std::remove(tmp_path("ho_bp4.log").c_str());
ix.open(tmp_path("ho_bp4.log"));
Protocol *alice = holder(&ix);
Wire w("w");
alice->add_peer(&w);
ask_for_list(*alice, w);
w.sent.clear();
msg::ChunkCancel cancel;
cancel.path = content_path(oid);
cancel.indices = { 1, 3 };
alice->on_message(&w, MsgId::ChunkCancel, msg::encode(cancel));
ask_for_chunks(*alice, w, { 0, 1, 2, 3, 4 });
const std::vector<uint64_t> got = w.data_indices();
check(got == std::vector<uint64_t>({ 0, 2, 4 }), "the cancelled chunks are skipped");
// One-shot: a later request for the same chunk is served normally, because
// the cancel referred to one assignment, not to the chunk forever.
w.sent.clear();
ask_for_chunks(*alice, w, { 1, 3 });
check(w.data_indices() == std::vector<uint64_t>({ 1, 3 }),
"and the cancel is one-shot — asking again is served");
delete alice;
}
printf("\n %d assertions pass, %d fail\n", passed, failed);
printf("%s\n", failed == 0 ? "PASS" : "FAIL");
return failed == 0 ? 0 : 1;
}