// Protocol driver: peers over an in-memory loopback, mirroring the JS reference
// harness (two overlays over a duplex pair).
//
// The assertions are the ones live-transcripts.json records as CONDUCT, not just
// "the transfer worked":
//
// fetch-by-hash contentRequest -> chunkHashes -> chunkNeed -> chunkData...
// not-held a request for content nobody has produces SILENCE, not an
// error frame
// lying holder bytes that do not match the advertised hash never land
// wire path the synthetic path is "content:"+hex, per the transcript
// multi-peer fan-out, first-holder-wins, per-peer serve authorisation
//
// Bytes live on DISK on both sides now: a holder serves from a file it never
// reads whole, and a requester assembles into a partial it never holds in memory.
// So these tests write real files and read the result back off disk.
//
// Build:
// g++ -O2 -std=c++17 -I liboverlay/include -I liboverlay/test \
// $(pkg-config --cflags libsodium) \
// -o /tmp/ho_proto liboverlay/test/test_protocol.cpp $(pkg-config --libs libsodium)
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <set>
#include <string>
#include <vector>
#include "overlay/protocol.h"
#include "json_mini.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 &name) { return std::string(TMP) + name; }
static std::string write_temp(const std::string &name, const std::vector<uint8_t> &data) {
const std::string p = tmp_path(name);
FILE *f = std::fopen(p.c_str(), "wb");
if (f) { std::fwrite(data.data(), 1, data.size(), f); std::fclose(f); }
return p;
}
static std::vector<uint8_t> read_file(const std::string &path) {
std::vector<uint8_t> out;
FILE *f = std::fopen(path.c_str(), "rb");
if (!f) return out;
std::fseek(f, 0, SEEK_END);
const long n = std::ftell(f);
std::fseek(f, 0, SEEK_SET);
if (n > 0) { out.resize((size_t)n); if (std::fread(out.data(), 1, out.size(), f) != out.size()) out.clear(); }
std::fclose(f);
return out;
}
// Loopback transport: hands frames straight to the other side and records the
// message names, so conduct can be asserted rather than inferred from outcomes.
struct Loopback : Transport {
Protocol *other = nullptr;
// The transport on the FAR side that points back at us. The receiving protocol
// is handed this as the message's `from`, mirroring the real adapter, where the
// OverlayTransport for a connection IS the peer identity.
Loopback *reverse = nullptr;
std::vector<std::string> *log = nullptr;
// Raw frames as sent, so a test can assert what actually went ON THE WIRE
// rather than what an internal field happened to hold. The wire-path check
// below needs exactly this distinction.
std::vector<std::pair<MsgId, std::vector<uint8_t>>> *frames = nullptr;
std::string label;
struct Item { Protocol *target; Transport *from; MsgId id; std::vector<uint8_t> frame; };
// Deliver synchronously but through a queue, so a handler that sends while
// handling does not recurse arbitrarily deep.
static inline std::vector<Item> queue;
// Backpressure simulation: after `writable_for` frames the carrier reports
// full, and stays that way until the test calls on_writable.
int writable_for = -1; // <0 = always writable
int sent_since_drain = 0;
bool send(MsgId id, const std::vector<uint8_t> &frame) override {
if (log) log->push_back(label + ":" + msg_name(id));
if (frames) frames->push_back({ id, frame });
queue.push_back({ other, reverse, id, frame });
if (writable_for < 0) return true;
return ++sent_since_drain < writable_for;
}
const char *peer_label() const override { return label.c_str(); }
};
// Wire two loopbacks into a connection: each knows the protocol it delivers to and
// the transport the far side sees it as. Adds rather than replaces, so a protocol
// can be linked to several peers.
static void connect(Loopback &a, Protocol &pa, Loopback &b, Protocol &pb) {
a.other = &pb; a.reverse = &b;
b.other = &pa; b.reverse = &a;
pa.add_peer(&a);
pb.add_peer(&b);
}
static int count_frames(const std::vector<std::pair<MsgId, std::vector<uint8_t>>> &f, MsgId want) {
int n = 0;
for (const auto &e : f) if (e.first == want) n++;
return n;
}
// Decode the `path` field of the first frame of type `want` in a capture.
static bool wire_path_of(const std::vector<std::pair<MsgId, std::vector<uint8_t>>> &frames,
MsgId want, std::string &out) {
for (const auto &f : frames) {
if (f.first != want) continue;
cenc::State s;
s.buffer = const_cast<uint8_t *>(f.second.data());
s.start = 0;
s.end = f.second.size();
out = cenc::string_decode(s); // `path` is the first field of all three
return true;
}
return false;
}
// Every chunk index this capture asked for, across all its chunkNeed frames.
static std::vector<uint64_t> wire_need_indices(const std::vector<std::pair<MsgId, std::vector<uint8_t>>> &frames) {
std::vector<uint64_t> out;
for (const auto &f : frames) {
if (f.first != MsgId::ChunkNeed) continue;
cenc::State s;
s.buffer = const_cast<uint8_t *>(f.second.data());
s.start = 0;
s.end = f.second.size();
const msg::ChunkNeed n = msg::decode_chunk_need(s);
out.insert(out.end(), n.indices.begin(), n.indices.end());
}
return out;
}
// Frames are queued globally, and every test block builds its own protocols on the
// stack. A block that ends with frames still queued would have them delivered into
// the NEXT block, against objects that no longer exist — a use-after-scope that
// presents as a crash somewhere unrelated. So each block starts clean, and says so.
static void reset_wire() { Loopback::queue.clear(); }
static size_t queued() { return Loopback::queue.size(); }
static void pump() {
// Drain until quiet. Real transports are async; this keeps the test
// deterministic without pretending the protocol is synchronous.
for (int guard = 0; guard < 100000 && !Loopback::queue.empty(); guard++) {
auto item = Loopback::queue.front();
Loopback::queue.erase(Loopback::queue.begin());
item.target->on_message(item.from, item.id, item.frame);
}
}
// Deliver at most `n` frames, so a test can stop a transfer half-way. pump()
// drains to quiet, which is the wrong instrument for anything that happens
// DURING a transfer.
static void pump_n(size_t n) {
for (size_t i = 0; i < n && !Loopback::queue.empty(); i++) {
auto item = Loopback::queue.front();
Loopback::queue.erase(Loopback::queue.begin());
item.target->on_message(item.from, item.id, item.frame);
}
}
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;
}
static std::string hash_of(const std::vector<uint8_t> &v) {
uint8_t h[32];
cdc::hash_chunk(v.data(), v.size(), h);
return to_hex(h, 32);
}
// A holder whose bytes are a real file. Returning a LOCATION rather than bytes is
// the contract now — the file may be larger than RAM.
static Protocol::ByteSource holder_of(const std::string &oid, const std::string &path) {
return [oid, path](const std::string &h, ContentLocation &out) {
if (h != oid) return false;
out.path = path;
return true; // size resolved by stat
};
}
static bool holds_nothing(const std::string &, ContentLocation &) { return false; }
int main() {
printf("hyper-overlay C++ protocol driver\n");
// ---- 1. fetch by content hash ------------------------------------------
{
printf("\n-- fetch-by-hash\n");
reset_wire();
std::vector<std::string> log;
const std::vector<uint8_t> content = make_bytes(200 * 1024, 7); // ~12 chunks
const std::string oid = hash_of(content);
const std::string src = write_temp("ho_p_src1.bin", content);
const std::string dest = tmp_path("ho_p_dst1.bin");
std::remove(dest.c_str());
std::remove(xfer::partial_path_for(dest).c_str());
index::FileIndex ia, ib;
std::remove(tmp_path("ho_p_a.log").c_str()); std::remove(tmp_path("ho_p_b.log").c_str());
ia.open(tmp_path("ho_p_a.log"));
ib.open(tmp_path("ho_p_b.log"));
Protocol alice(&ia, holder_of(oid, src));
Protocol bob(&ib, holds_nothing);
alice.set_local_label("alice"); bob.set_local_label("bob");
Loopback la, lb;
la.log = &log; la.label = "alice";
lb.log = &log; lb.label = "bob";
connect(la, alice, lb, bob);
bool done = false;
std::string got_path;
bob.on_complete = [&](const std::string &, const std::string &p) { got_path = p; done = true; };
bob.request(oid, dest);
pump();
check(done, "transfer completed");
const std::vector<uint8_t> received = read_file(dest);
check(received.size() == content.size(), "received length matches");
check(received == content, "received bytes reconstruct the original exactly");
check(got_path == dest, "completion reports the file's path, not its bytes");
// The partial is the transfer's working file and must not survive it.
check(!io::file_exists(xfer::partial_path_for(dest)), "the partial is renamed away, not left behind");
// Conduct, in the order the JS transcript records.
auto idx_of = [&](const std::string &n) -> int {
for (size_t i = 0; i < log.size(); i++) if (log[i] == n) return (int)i;
return -1;
};
const int req = idx_of("bob:contentRequest");
const int hashes = idx_of("alice:chunkHashes");
const int need = idx_of("bob:chunkNeed");
const int data = idx_of("alice:chunkData");
check(req == 0, "requester opens with contentRequest");
check(hashes > req, "holder answers with chunkHashes");
check(need > hashes, "requester follows with chunkNeed");
check(data > need, "holder then sends chunkData");
int datas = 0;
for (const std::string &e : log) if (e == "alice:chunkData") datas++;
check(datas > 1, "a multi-chunk file sends multiple chunkData frames (" +
std::to_string(datas) + ")");
check(idx_of("alice:contentRequest") < 0, "the holder never sends a request back");
}
// ---- 2. a request nobody can serve is SILENCE --------------------------
{
printf("\n-- unheld content\n");
reset_wire();
std::vector<std::string> log;
index::FileIndex ia, ib;
std::remove(tmp_path("ho_p_c.log").c_str()); std::remove(tmp_path("ho_p_d.log").c_str());
ia.open(tmp_path("ho_p_c.log"));
ib.open(tmp_path("ho_p_d.log"));
Protocol alice(&ia, holds_nothing);
Protocol bob(&ib, holds_nothing);
Loopback la, lb;
la.log = &log; la.label = "alice";
lb.log = &log; lb.label = "bob";
connect(la, alice, lb, bob);
sched::Options so; so.idle_timeout_ms = 500;
bob.configure_scheduler(so);
std::string failure;
bob.on_failed = [&](const std::string &, const std::string &why) { failure = why; };
bob.tick(0);
const std::string missing(64, 'a');
bob.request(missing, tmp_path("ho_p_never.bin"));
pump();
check(log.size() == 1 && log[0] == "bob:contentRequest",
"an unservable request draws no reply at all (no error frame)");
// Silence is the correct ANSWER, but it must not be an infinite wait. Nothing
// else can end this fetch: no holder answered, so no chunk list arrives, no
// chunk is ever rejected, and no peer disconnects. Only a watchdog armed when
// the REQUEST went out will ever fire — which is why it is armed there and
// not on the first response.
bob.tick(400);
check(failure.empty() && bob.has_download(missing), "the fetch stays open inside the window");
bob.tick(501);
check(failure.find("stalled") != std::string::npos,
"and past it, a request nobody answered is abandoned (" + failure + ")");
check(!bob.has_download(missing), "the download is dropped");
}
// ---- 3. corrupt bytes from a peer must not reach the file --------------
//
// The overlay has NO serve authorisation — anyone on the topic can answer —
// so a holder serving wrong bytes is a case that must be handled, not an
// impossibility.
{
printf("\n-- a lying holder\n");
reset_wire();
std::vector<std::string> log;
const std::vector<uint8_t> content = make_bytes(64 * 1024, 11);
const std::string oid = hash_of(content);
// Alice advertises the right hashes but her file holds DIFFERENT bytes.
std::vector<uint8_t> lies = content;
for (size_t i = 0; i < lies.size(); i++) lies[i] ^= 0xff;
const std::string src = write_temp("ho_p_lies.bin", lies);
const std::string dest = tmp_path("ho_p_dst3.bin");
std::remove(dest.c_str());
std::remove(xfer::partial_path_for(dest).c_str());
index::FileIndex ia, ib;
std::remove(tmp_path("ho_p_e.log").c_str()); std::remove(tmp_path("ho_p_f.log").c_str());
ia.open(tmp_path("ho_p_e.log"));
ib.open(tmp_path("ho_p_f.log"));
// Alice's index is pre-seeded with the chunk map of the TRUE content, so
// she advertises honest hashes and then serves bytes that do not match
// them. That is the actual attack; an earlier version of this test toggled
// a byte source on a flag and proved nothing, because the serve path reads
// the same file for both the advertisement and the bytes.
{
const std::vector<cdc::Chunk> truth = cdc::chunk(content.data(), content.size());
std::vector<index::ChunkRef> refs;
for (const cdc::Chunk &c : truth)
refs.push_back({ to_hex(c.hash, 32), c.offset, c.length });
ia.put_chunk_map_by_hash(oid, refs);
}
Protocol alice(&ia, holder_of(oid, src));
Protocol bob(&ib, holds_nothing);
Loopback la, lb;
la.log = &log; la.label = "alice";
lb.log = &log; lb.label = "bob";
connect(la, alice, lb, bob);
bool done = false;
std::string failure;
bob.on_complete = [&](const std::string &, const std::string &) { done = true; };
bob.on_failed = [&](const std::string &, const std::string &why) { failure = why; };
// A holder that lies CONSISTENTLY is not self-limiting: every rejected chunk
// is offered to a peer again, so without a watchdog the fetch spins forever
// re-requesting bytes that will never verify. The idle watchdog is what ends
// it — rejections deliberately do not count as progress.
sched::Options so; so.idle_timeout_ms = 1000; so.cap = 4;
bob.configure_scheduler(so);
bob.tick(0);
bob.request(oid, dest);
pump();
check(!done, "a transfer serving wrong bytes never completes");
const Download *d = bob.download(oid);
check(d && d->transfer.received() == 0, "not one corrupted chunk was accepted");
check(!io::file_exists(dest), "and nothing was written under the destination name");
bob.tick(2000);
check(failure.find("stalled") != std::string::npos,
"and the fetch is eventually abandoned rather than retried forever (" + failure + ")");
check(!bob.has_download(oid), "the download is dropped");
reset_wire();
}
// ---- 4. the synthetic path is wire-visible, and not ours to choose ------
//
// chunkHashes/chunkNeed/chunkData all carry a `path`, and the reference sets it
// to "content:" + hex. Keying it by the bare hash is SELF-CONSISTENT — every
// test above passes either way, because both sides of a loopback share whatever
// convention we picked — and cannot move a byte to or from the JS peers running
// in production. So this asserts against the recorded transcript, not against
// the other end of our own wire.
{
printf("\n-- wire path format\n");
reset_wire();
std::vector<std::pair<MsgId, std::vector<uint8_t>>> a_frames, b_frames;
const std::vector<uint8_t> content = make_bytes(200 * 1024, 33);
const std::string oid = hash_of(content);
const std::string src = write_temp("ho_p_src4.bin", content);
const std::string dest = tmp_path("ho_p_dst4.bin");
std::remove(dest.c_str());
std::remove(xfer::partial_path_for(dest).c_str());
index::FileIndex ia, ib;
std::remove(tmp_path("ho_p_g.log").c_str()); std::remove(tmp_path("ho_p_h.log").c_str());
ia.open(tmp_path("ho_p_g.log"));
ib.open(tmp_path("ho_p_h.log"));
Protocol alice(&ia, holder_of(oid, src));
Protocol bob(&ib, holds_nothing);
Loopback la, lb;
la.frames = &a_frames; la.label = "alice";
lb.frames = &b_frames; lb.label = "bob";
connect(la, alice, lb, bob);
bob.request(oid, dest);
pump();
std::string hashes_path, need_path, data_path;
check(wire_path_of(a_frames, MsgId::ChunkHashes, hashes_path), "holder sent a chunkHashes frame");
check(wire_path_of(b_frames, MsgId::ChunkNeed, need_path), "requester sent a chunkNeed frame");
check(wire_path_of(a_frames, MsgId::ChunkData, data_path), "holder sent a chunkData frame");
check(hashes_path == "content:" + oid,
"chunkHashes.path is \"content:\"+hex, not the bare hash");
check(need_path == hashes_path && data_path == hashes_path,
"chunkNeed and chunkData echo the same synthetic path");
// Pin it to the fixture rather than to this test's belief about the fixture.
const char *conf = getenv("HO_CONFORMANCE");
const std::string tpath = std::string(conf ? conf
: "../../hypercore-kb/hyper-fleeky/hyper-overlay/test/conformance") + "/live-transcripts.json";
std::string raw;
if (!read_text_file(tpath, raw)) {
printf(" SKIP live-transcripts.json not found at %s\n", tpath.c_str());
} else {
JsonParser jp{ raw.c_str() };
const Json root = jp.parse();
std::string ref_path;
const Json *ts = root.find("transcripts");
const Json *fbh = ts ? ts->find("fetch-by-hash") : nullptr;
if (fbh) {
for (const Json &f : fbh->arr) {
const Json *n = f.find("name");
if (!n || n->str != "chunkHashes") continue;
const Json *sum = f.find("summary");
const Json *pth = sum ? sum->find("path") : nullptr;
if (pth) ref_path = pth->str;
break;
}
}
check(!ref_path.empty(), "transcript records a chunkHashes path to compare against");
// Same prefix, same shape (prefix + 64 hex) — the hashes differ because
// the payloads differ, which is why this compares form and not value.
const std::string pfx = "content:";
check(ref_path.rfind(pfx, 0) == 0 && ref_path.size() == pfx.size() + 64,
"the JS reference keys it \"content:\"+64hex (" + ref_path.substr(0, 16) + "...)");
check(hashes_path.rfind(pfx, 0) == 0 && hashes_path.size() == ref_path.size(),
"our frame matches the reference's key format exactly");
}
}
// ---- 5. several peers on one protocol ----------------------------------
//
// The driver is multi-peer. Multi-SOURCE (splitting a file's chunks across
// holders) is the transfer manager's job and deliberately absent, so what must
// hold today is: the request reaches everyone, exactly one holder ends up
// serving it, and a peer cannot pull bytes it never asked for.
{
printf("\n-- multi-peer\n");
reset_wire();
std::vector<std::pair<MsgId, std::vector<uint8_t>>> f_bob_a, f_bob_c, f_alice_c;
const std::vector<uint8_t> content = make_bytes(200 * 1024, 55);
const std::string oid = hash_of(content);
const std::string src = write_temp("ho_p_src5.bin", content);
const std::string dest = tmp_path("ho_p_dst5.bin");
std::remove(dest.c_str());
std::remove(xfer::partial_path_for(dest).c_str());
index::FileIndex ia, ib, ic;
for (const char *p : { "ho_p_i.log", "ho_p_j.log", "ho_p_k.log" }) std::remove(tmp_path(p).c_str());
ia.open(tmp_path("ho_p_i.log"));
ib.open(tmp_path("ho_p_j.log"));
ic.open(tmp_path("ho_p_k.log"));
Protocol alice(&ia, holder_of(oid, src)); // holder
Protocol carol(&ic, holder_of(oid, src)); // holder too
Protocol bob(&ib, holds_nothing);
// bob is connected to BOTH holders; alice additionally sees carol.
Loopback la_b, lb_a, lc_b, lb_c, la_c, lc_a;
la_b.label = "alice"; lb_a.label = "bob"; lb_a.frames = &f_bob_a;
lc_b.label = "carol"; lb_c.label = "bob"; lb_c.frames = &f_bob_c;
la_c.label = "alice"; lc_a.label = "carol"; la_c.frames = &f_alice_c;
connect(la_b, alice, lb_a, bob);
connect(lc_b, carol, lb_c, bob);
connect(la_c, alice, lc_a, carol);
check(bob.peer_count() == 2, "requester holds both connections");
check(alice.peer_count() == 2, "a holder can be connected to several peers");
int completions = 0;
bob.on_complete = [&](const std::string &, const std::string &) { completions++; };
// A small window so the work is actually SPLIT. With the default window one
// holder would take every chunk in the first assignment and the second would
// have nothing left to contribute — true, but it would prove nothing about
// multi-source.
sched::Options so; so.cap = 2; so.hotswap_stall_ms = -1;
bob.configure_scheduler(so);
bob.tick(0);
bob.request(oid, dest);
pump();
check(count_frames(f_bob_a, MsgId::ContentRequest) == 1 &&
count_frames(f_bob_c, MsgId::ContentRequest) == 1,
"the request fans out to EVERY connected peer");
check(completions == 1 && read_file(dest) == content, "the file arrives exactly once, intact");
// Both holders answer, and BOTH are used — that is the point of multi-source.
// What must not happen is the same chunk being pulled from both, which is
// the bandwidth the whole layer exists to save.
const std::vector<uint64_t> from_a = wire_need_indices(f_bob_a);
const std::vector<uint64_t> from_c = wire_need_indices(f_bob_c);
check(!from_a.empty() && !from_c.empty(),
"the work is split across both holders (" + std::to_string(from_a.size()) +
" + " + std::to_string(from_c.size()) + " chunks requested)");
std::set<uint64_t> asked;
bool twice = false;
for (uint64_t i : from_a) if (!asked.insert(i).second) twice = true;
for (uint64_t i : from_c) if (!asked.insert(i).second) twice = true;
check(!twice, "and no chunk is ever requested from more than one holder");
// A peer that never asked must not be able to pull bytes by sending
// chunkNeed for a path some OTHER peer's request registered.
const size_t before = f_alice_c.size();
msg::ChunkNeed poach;
poach.path = content_path(oid);
poach.indices = { 0, 1 };
lc_a.send(MsgId::ChunkNeed, msg::encode(poach));
pump();
check(count_frames(f_alice_c, MsgId::ChunkData) == 0,
"a peer that never requested gets no bytes for another peer's path");
check(f_alice_c.size() == before, "and draws no reply at all");
bob.remove_peer(&lb_a);
check(bob.peer_count() == 1, "a departing peer is dropped");
}
// ---- 6. paging a chunk list ---------------------------------------------
//
// A chunk list for a large file does not fit in one frame, and the transport
// does not truncate it — @hyperswarm/secret-stream REFUSES an atomic write over
// 16 MiB-1 and drops the connection, so the transfer dies before a byte of
// content ships.
{
printf("\n-- wire paging\n");
reset_wire();
// The page size exists to stay under the transport limit. Assert the
// relationship rather than leaving it as arithmetic in a comment.
{
msg::ChunkHashes probe;
probe.path = content_path(std::string(64, 'a'));
probe.tier = 3;
probe.chunks.resize(MAX_CHUNKS_PER_MSG);
for (auto &c : probe.chunks) c.length = 4u * 1024u * 1024u;
const size_t bytes = msg::encode(probe).size();
check(MAX_CHUNKS_PER_MSG == 100000, "the page size is the reference's 100k entries");
check(bytes < NOISE_MAX_ATOMIC_WRITE,
"a full page encodes to " + std::to_string(bytes / 1024) + " KiB, under the " +
std::to_string(NOISE_MAX_ATOMIC_WRITE / 1024 / 1024) + " MiB atomic-write limit");
}
const std::vector<uint8_t> content = make_bytes(200 * 1024, 77);
const std::string oid = hash_of(content);
const std::string src = write_temp("ho_p_src6.bin", content);
const std::string dest = tmp_path("ho_p_dst6.bin");
const size_t true_chunks = cdc::chunk(content.data(), content.size()).size();
// A list that FITS ships as exactly one frame — the small-file norm, and
// identical on the wire to the unpaged format bar a trailing zero.
{
std::remove(dest.c_str()); std::remove(xfer::partial_path_for(dest).c_str());
reset_wire();
std::vector<std::pair<MsgId, std::vector<uint8_t>>> af;
index::FileIndex ia, ib;
std::remove(tmp_path("ho_p_p1.log").c_str()); std::remove(tmp_path("ho_p_p2.log").c_str());
ia.open(tmp_path("ho_p_p1.log")); ib.open(tmp_path("ho_p_p2.log"));
Protocol alice(&ia, holder_of(oid, src));
Protocol bob(&ib, holds_nothing);
Loopback la, lb; la.frames = ⁡ la.label = "alice"; lb.label = "bob";
connect(la, alice, lb, bob);
bool done = false;
bob.on_complete = [&](const std::string &, const std::string &) { done = true; };
bob.request(oid, dest);
pump();
check(count_frames(af, MsgId::ChunkHashes) == 1, "a list that fits ships as one frame");
check(done && read_file(dest) == content, "and the file arrives intact");
}
// Forced paging: same file, tiny page size.
{
std::remove(dest.c_str()); std::remove(xfer::partial_path_for(dest).c_str());
reset_wire();
std::vector<std::pair<MsgId, std::vector<uint8_t>>> af;
index::FileIndex ia, ib;
std::remove(tmp_path("ho_p_p3.log").c_str()); std::remove(tmp_path("ho_p_p4.log").c_str());
ia.open(tmp_path("ho_p_p3.log")); ib.open(tmp_path("ho_p_p4.log"));
Protocol alice(&ia, holder_of(oid, src));
Protocol bob(&ib, holds_nothing);
alice.set_max_chunks_per_message(3);
Loopback la, lb; la.frames = ⁡ la.label = "alice"; lb.label = "bob";
connect(la, alice, lb, bob);
bool done = false;
bob.on_complete = [&](const std::string &, const std::string &) { done = true; };
bob.request(oid, dest);
pump();
const size_t expect_frames = (true_chunks + 2) / 3;
check(count_frames(af, MsgId::ChunkHashes) == (int)expect_frames,
std::to_string(true_chunks) + " chunks at 3 per page ship as " +
std::to_string(expect_frames) + " frames");
// Every frame but the last must say more:1, or the receiver dispatches an
// incomplete list and asks for a file that is missing its tail.
int more_set = 0;
for (const auto &f : af) {
if (f.first != MsgId::ChunkHashes) continue;
cenc::State st; st.buffer = const_cast<uint8_t *>(f.second.data());
st.start = 0; st.end = f.second.size();
if (msg::decode_chunk_hashes(st).more) more_set++;
}
check(more_set == (int)expect_frames - 1, "every page but the last carries more:1");
check(done && read_file(dest) == content,
"and the reassembled list drives a byte-identical transfer");
check(bob.buffered_page_entries() == 0, "no page fragments are left held afterwards");
}
// F2: a holder that pages forever must not grow our memory without bound.
// It cannot be stopped by the stall watchdog either, because every buffered
// page counts as progress — the two would cover for each other indefinitely.
{
std::remove(dest.c_str()); std::remove(xfer::partial_path_for(dest).c_str());
reset_wire();
std::vector<std::pair<MsgId, std::vector<uint8_t>>> bf;
index::FileIndex ib;
std::remove(tmp_path("ho_p_p5.log").c_str());
ib.open(tmp_path("ho_p_p5.log"));
Protocol bob(&ib, holds_nothing);
bob.set_max_chunk_hashes_accum(8);
Loopback lb; lb.frames = &bf; lb.label = "bob";
bob.add_peer(&lb);
bob.request(oid, dest);
reset_wire();
// Stream pages that never end.
msg::ChunkHashes page;
page.path = content_path(oid);
page.tier = 0;
page.chunks.resize(3);
page.more = 1;
for (int i = 0; i < 50; i++)
bob.on_message(&lb, MsgId::ChunkHashes, msg::encode(page));
check(bob.buffered_page_entries() <= 8,
"the accumulator is capped, not unbounded (" +
std::to_string(bob.buffered_page_entries()) + " entries held after 150)");
check(count_frames(bf, MsgId::ChunkNeed) == 0, "no chunk is ever requested from it");
check(bob.has_download(oid), "the fetch is still open for another holder to answer");
reset_wire();
}
// Pages for a path we never asked for are dropped, not buffered.
{
reset_wire();
index::FileIndex ib;
std::remove(tmp_path("ho_p_p6.log").c_str());
ib.open(tmp_path("ho_p_p6.log"));
Protocol bob(&ib, holds_nothing);
Loopback lb; lb.label = "bob";
bob.add_peer(&lb);
msg::ChunkHashes page;
page.path = content_path(std::string(64, 'b'));
page.chunks.resize(100);
page.more = 1;
for (int i = 0; i < 20; i++) bob.on_message(&lb, MsgId::ChunkHashes, msg::encode(page));
check(bob.buffered_page_entries() == 0,
"a peer cannot make us hold pages for a path we are not fetching");
reset_wire();
}
// A paged list can take longer to stream than the idle window. Buffering a
// page is progress, or a big transfer is declared stalled before it starts.
{
std::remove(dest.c_str()); std::remove(xfer::partial_path_for(dest).c_str());
reset_wire();
index::FileIndex ib;
std::remove(tmp_path("ho_p_p7.log").c_str());
ib.open(tmp_path("ho_p_p7.log"));
Protocol bob(&ib, holds_nothing);
sched::Options so; so.idle_timeout_ms = 1000;
bob.configure_scheduler(so);
std::string failure;
bob.on_failed = [&](const std::string &, const std::string &why) { failure = why; };
Loopback lb; lb.label = "bob";
bob.add_peer(&lb);
bob.tick(0);
bob.request(oid, dest);
const std::vector<cdc::Chunk> all = cdc::chunk(content.data(), content.size());
uint64_t now = 0;
for (size_t i = 0; i < all.size(); i++) {
msg::ChunkHashes page;
page.path = content_path(oid);
page.tier = 0;
msg::ChunkInfo ci;
std::memcpy(ci.hash.data(), all[i].hash, 32);
ci.length = all[i].length;
page.chunks.push_back(ci);
page.more = (i + 1 < all.size()) ? 1 : 0;
now += 800; // slower than the 1000ms idle window allows twice over
bob.tick(now);
bob.on_message(&lb, MsgId::ChunkHashes, msg::encode(page));
}
check(failure.empty(),
"a chunk list streamed slower than the idle window does not trip the watchdog");
check(bob.has_download(oid), "the fetch survives to receive its list");
reset_wire();
}
// A fetch that dies mid-list must not leave its fragments behind: the next
// attempt at the same hash reuses the same key, and appending fresh pages
// onto a stale list would build a chunk map that never existed.
{
std::remove(dest.c_str()); std::remove(xfer::partial_path_for(dest).c_str());
reset_wire();
index::FileIndex ib;
std::remove(tmp_path("ho_p_p8.log").c_str());
ib.open(tmp_path("ho_p_p8.log"));
Protocol bob(&ib, holds_nothing);
sched::Options so; so.idle_timeout_ms = 100;
bob.configure_scheduler(so);
std::string failure;
bob.on_failed = [&](const std::string &, const std::string &why) { failure = why; };
Loopback lb; lb.label = "bob";
bob.add_peer(&lb);
bob.tick(0);
bob.request(oid, dest);
msg::ChunkHashes page;
page.path = content_path(oid);
page.chunks.resize(4);
page.more = 1;
bob.on_message(&lb, MsgId::ChunkHashes, msg::encode(page));
check(bob.buffered_page_entries() == 4, "a half-received list is held while it streams");
// The list never finishes; the holder goes away.
bob.remove_peer(&lb);
check(!failure.empty(), "losing the only holder ends the fetch (" + failure + ")");
check(bob.buffered_page_entries() == 0, "and its page fragments are dropped with it");
reset_wire();
}
}
// ---- 7. a holder that attaches AFTER the request ------------------------
//
// Every other block in this file connects its peers and then requests. Layer 12
// showed that is the UNUSUAL ordering in the field: the peers mesh in 0.1s and
// the overlay channel opens seconds later, so a fetch issued on connect fans out
// to nobody. A holder that arrives afterwards was never asked, and silence is
// its correct answer to a question it did not receive (rule 1) — so the fetch
// sits until the watchdog kills it with a willing holder attached. That was
// measured as 8.8s per fetch before it was measured as a stall.
//
// Hence the re-send in add_peer. Negative control, run: delete that loop and 3
// of this block's 6 assertions fail — the holder sees no contentRequest and the
// file never arrives — while all 57 other assertions in this file still pass,
// which is exactly why the suite could not see the gap.
{
printf("\n-- a holder that joins mid-fetch\n");
reset_wire();
std::vector<std::pair<MsgId, std::vector<uint8_t>>> f_bob_a;
const std::vector<uint8_t> content = make_bytes(200 * 1024, 77);
const std::string oid = hash_of(content);
const std::string src = write_temp("ho_p_src7.bin", content);
const std::string dest = tmp_path("ho_p_dst7.bin");
std::remove(dest.c_str());
std::remove(xfer::partial_path_for(dest).c_str());
index::FileIndex ia, ib;
for (const char *p : { "ho_p_q.log", "ho_p_r.log" }) std::remove(tmp_path(p).c_str());
ia.open(tmp_path("ho_p_q.log"));
ib.open(tmp_path("ho_p_r.log"));
Protocol alice(&ia, holder_of(oid, src));
Protocol bob(&ib, holds_nothing);
alice.set_local_label("alice"); bob.set_local_label("bob");
bool done = false;
bob.on_complete = [&](const std::string &, const std::string &) { done = true; };
bob.tick(0);
// The fetch begins with NOBODY connected — the layer 12 ordering.
bob.request(oid, dest);
check(bob.peer_count() == 0 && queued() == 0,
"a request with no peers connected puts nothing on the wire");
check(bob.has_download(oid), "and stays open, waiting for someone to ask");
Loopback la, lb;
la.label = "alice"; lb.label = "bob"; lb.frames = &f_bob_a;
connect(la, alice, lb, bob);
check(count_frames(f_bob_a, MsgId::ContentRequest) == 1,
"the holder is asked the moment it attaches, without a new request()");
// Attaching is driven by a channel callback, and a callback that fires twice
// must not double the fan-out.
bob.add_peer(&lb);
check(count_frames(f_bob_a, MsgId::ContentRequest) == 1,
"and re-adding the same peer does not ask it twice");
pump();
check(done && read_file(dest) == content,
"the fetch recovers and the file arrives intact");
// The re-send is per in-flight download, so a peer arriving with nothing
// outstanding must cost a frame of nothing.
std::vector<std::pair<MsgId, std::vector<uint8_t>>> f_late;
Loopback l_late; l_late.label = "late"; l_late.frames = &f_late;
bob.add_peer(&l_late);
check(f_late.empty() && queued() == 0,
"a peer joining with no fetch in flight is sent nothing");
reset_wire();
}
// ---- a contentRequest that arrives BEFORE serve() lands ----------------
// THE RACE THIS EXISTS FOR. HyperOverlay::serve() marshals onto the UDX thread
// and builds the chunk map there, so a hash is NOT resolvable the instant the
// game calls serve(). A peer that asks in that window used to be answered with
// SILENCE and never asked again — costing the requester the full 30s idle
// watchdog for a file the holder was about to have. Observed for real: three of
// four concurrent fetches failed with `source resolved=NO`, and layer 13 had to
// paper over it with a settle-pump.
//
// The wire must stay silent (rule 1: no "not found" frame exists, and inventing
// one would have every peer answer every request), so the retry is LOCAL: park
// the unresolved request and re-attempt it on tick.
{
printf("\n-- a request arriving before the holder is ready\n");
reset_wire();
std::vector<std::string> log;
const std::vector<uint8_t> content = make_bytes(200 * 1024, 11);
const std::string oid = hash_of(content);
const std::string src = write_temp("ho_late_src.bin", content);
const std::string dest = tmp_path("ho_late_dst.bin");
std::remove(dest.c_str());
std::remove(xfer::partial_path_for(dest).c_str());
index::FileIndex ia, ib;
std::remove(tmp_path("ho_late_a.log").c_str()); std::remove(tmp_path("ho_late_b.log").c_str());
ia.open(tmp_path("ho_late_a.log")); ib.open(tmp_path("ho_late_b.log"));
// serve() has NOT landed yet: the holder resolves nothing.
bool served = false;
Protocol alice(&ia, [&](const std::string &h, ContentLocation &out) {
if (!served || h != oid) return false;
out.path = src;
return true;
});
Protocol bob(&ib, holds_nothing);
alice.set_local_label("alice"); bob.set_local_label("bob");
Loopback la, lb;
la.log = &log; la.label = "alice";
lb.log = &log; lb.label = "bob";
connect(la, alice, lb, bob);
bool done = false;
bob.on_complete = [&](const std::string &, const std::string &) { done = true; };
bob.request(oid, dest);
pump();
check(!done, "nothing transfers while the holder cannot resolve it");
bool answered = false;
for (const std::string &e : log) if (e == "alice:chunkHashes") answered = true;
check(!answered, "and the holder stays SILENT on the wire, as rule 1 requires");
// serve() lands. Nothing new arrives on the wire — the holder must notice
// by itself that it can now answer a request it already has.
served = true;
alice.tick(1000);
pump();
check(done, "the parked request is answered once serve() lands");
const std::vector<uint8_t> received = read_file(dest);
check(received == content, "and the bytes are the original");
}
// ---- cancel, and what a caller can see while a transfer runs -----------
//
// A fetch that cannot be stopped is a fetch that owns the disk until it
// finishes, and a caller with no view of it can only report "downloading".
// swarmd's CONTENT_CANCEL and CONTENT_PROGRESS are these two methods.
//
// The distinction the partial exists for: pausing KEEPS it so the next
// request resumes; discarding removes it.
{
printf("\n-- cancel and progress\n");
reset_wire();
const std::vector<uint8_t> content = make_bytes(200 * 1024, 23);
const std::string oid = hash_of(content);
const std::string src = write_temp("ho_p_src_cancel.bin", content);
const std::string dest = tmp_path("ho_p_dst_cancel.bin");
std::remove(dest.c_str());
std::remove(xfer::partial_path_for(dest).c_str());
index::FileIndex ia, ib;
std::remove(tmp_path("ho_p_ca.log").c_str()); std::remove(tmp_path("ho_p_cb.log").c_str());
ia.open(tmp_path("ho_p_ca.log"));
ib.open(tmp_path("ho_p_cb.log"));
Protocol alice(&ia, holder_of(oid, src));
Protocol bob(&ib, holds_nothing);
Loopback la, lb;
la.label = "alice"; lb.label = "bob";
connect(la, alice, lb, bob);
bool done = false;
bob.on_complete = [&](const std::string &, const std::string &) { done = true; };
uint64_t bytes = 0; size_t got = 0, total = 0;
check(!bob.progress(oid, bytes, got, total),
"progress on a hash nobody is fetching is not an answer");
bob.request(oid, dest);
pump_n(4); // request -> chunkHashes -> chunkNeed -> some chunkData
check(bob.progress(oid, bytes, got, total), "progress is readable mid-transfer");
check(total > 1, "and knows how many chunks the file has (" + std::to_string(total) + ")");
check(got > 0 && got < total,
"and reports a transfer in progress, not 0 and not finished (" +
std::to_string(got) + "/" + std::to_string(total) + ")");
check(bob.cancel(oid, /*discard=*/false), "cancel finds the download");
check(!bob.has_download(oid), "and the download is gone");
check(io::file_exists(xfer::partial_path_for(dest)),
"a PAUSED cancel keeps the partial — that is what it is for");
pump(); // frames still in flight must land nowhere
check(!done, "nothing completes after a cancel");
check(!io::file_exists(dest), "and no destination file appears");
// ---- and the same fetch resumes rather than restarting ----
bob.request(oid, dest);
pump();
check(done, "the fetch resumes and completes");
check(read_file(dest) == content, "and the resumed bytes are the original");
// ---- discard removes the partial ----
const std::string dest2 = tmp_path("ho_p_dst_cancel2.bin");
std::remove(dest2.c_str());
std::remove(xfer::partial_path_for(dest2).c_str());
bob.request(oid, dest2);
pump_n(4);
check(bob.cancel(oid, /*discard=*/true), "cancel finds the second download");
check(!io::file_exists(xfer::partial_path_for(dest2)),
"a DISCARDING cancel takes the partial with it");
check(!bob.cancel(oid, false), "cancelling what is not running is false, not a crash");
}
printf("\n %d assertions pass, %d fail\n", passed, failed);
printf("%s\n", failed == 0 ? "PASS" : "FAIL");
return failed == 0 ? 0 : 1;
}