#ifndef HYPEROVERLAY_PROTOCOL_H
#define HYPEROVERLAY_PROTOCOL_H
// The fetch/serve state machine: request a file by content hash from the peers
// that hold it, serve one to a peer who asks.
//
// SCOPE, stated up front so it is not mistaken for more than it is. This drives
// the flow the JS reference records in live-transcripts.json:
//
// contentRequest -> chunkHashes -> chunkNeed -> chunkData... -> (complete)
//
// It is MULTI-SOURCE: a request fans out to every connected peer, and every peer
// that answers with a matching chunk list joins the fetch. The scheduler
// (scheduler.h) decides who is asked for what; this file only moves frames and
// bytes. Wire paging and single-flight backpressure are still to come — see
// DESIGN-transfer-manager.md.
//
// NEITHER SIDE HOLDS A WHOLE FILE. A holder answers with a PATH, not with bytes,
// and reads each chunk at its offset when asked for it. A requester writes into a
// partial through one handle and never assembles the file in memory (transfer.h).
// This is invisible from the wire — the frames are identical either way — which
// is why it is built deliberately rather than discovered when a large asset kills
// the process.
//
// TWO RULES THE TRANSCRIPTS ENCODE, both of which are easy to get wrong and
// expensive in the field:
//
// 1. A HOLDER SERVES ONLY WHAT IT HAS. On contentRequest the holder looks the
// hash up in its own index; not finding it is silence, not an error frame.
// Answering "no" to every request a peer makes turns a swarm into a
// broadcast storm.
// 2. A REQUESTER ASKS ONLY FOR WHAT IT LACKS. chunkNeed carries the indices
// still missing, computed against what is already on disk. Asking for
// everything works and is why resume exists as a concept — it just wastes
// the bandwidth the whole layer is for.
//
// Verification is per chunk, on arrival, against the hash the holder advertised —
// and the assembled file is verified against the content hash that was ASKED
// for. A peer that serves the wrong bytes is caught before they reach disk,
// which matters because the overlay has no serve authorisation: anyone on the
// topic can answer.
#include <algorithm>
#include <chrono>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <deque>
#include <functional>
#include <map>
#include <memory>
#include <set>
#include <string>
#include <vector>
#include "chunker.h"
#include "file_index.h"
#include "fileio.h"
#include "messages.h"
#include "scheduler.h"
#include "transfer.h"
#include "transport.h"
namespace hyperoverlay {
// Fetch-lifecycle tracing. QUIET BY DEFAULT — set HO_VERBOSE=1 to enable.
//
// This is the same discipline as hyperswarm-peer's own debug_log.h (which this
// library cannot include without breaking extraction, so the gate is duplicated
// rather than shared): output that costs nothing when off, and per-FRAME lines
// are not written at all. Four call sites remain, all of them once per fetch or
// once per peer — request fan-out, a peer joining mid-fetch, a contentRequest
// arriving, a serve starting. Those four are what localised the layer-12 defect
// (hyper-games #14 §5), which is why they survived the strip and the per-frame
// SEND/RECV lines did not.
//
// Timestamped because stderr is unbuffered while stdout is block-buffered through
// a pipe: their interleaved order proves nothing, and a run was misread for
// exactly that reason.
inline uint64_t ho_now_ms() {
return (uint64_t)std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now().time_since_epoch()).count();
}
// Evaluated once (C++11 thread-safe static init).
inline bool ho_log_enabled() {
static const bool enabled = std::getenv("HO_VERBOSE") != nullptr;
return enabled;
}
#define HO_LOG(...) do { if (hyperoverlay::ho_log_enabled()) { \
fprintf(stderr, "[overlay t=%llu] ", (unsigned long long)hyperoverlay::ho_now_ms()); \
fprintf(stderr, __VA_ARGS__); } } while (0)
// to_hex lives in chunker.h — the layer that produces the hashes it formats.
inline msg::Hash32 from_hex32(const std::string &h) {
msg::Hash32 out{};
out.fill(0);
for (size_t i = 0; i + 1 < h.size() && i / 2 < 32; i += 2)
out[i / 2] = (uint8_t)strtoul(h.substr(i, 2).c_str(), nullptr, 16);
return out;
}
// THE SYNTHETIC PATH — wire-visible, and not ours to choose.
//
// Every frame of a content-addressed fetch (chunkHashes, chunkNeed, chunkData)
// carries a `path`, and the reference sets it to "content:" + hex
// (lib/protocol.js:999). It is what the requester keys its in-flight download by
// and what the holder keys its serve state by, on BOTH sides of the wire.
//
// This layer originally used the bare hex hash. That is self-consistent — a C++
// peer talks to a C++ peer perfectly — and cannot exchange a single byte with the
// JS peers in production: their chunkHashes arrives keyed "content:<hex>", our
// lookup is keyed "<hex>", it misses, and the transfer stops with no error
// anywhere. Pinned to the recorded transcript in test_protocol.cpp so it cannot
// drift back.
static constexpr const char *CONTENT_PREFIX = "content:";
inline std::string content_path(const std::string &contentHash) { return std::string(CONTENT_PREFIX) + contentHash; }
inline bool is_content_path(const std::string &p) { return p.rfind(CONTENT_PREFIX, 0) == 0; }
inline std::string content_hash_of(const std::string &p) { return is_content_path(p) ? p.substr(8) : p; }
// ---------------------------------------------------------------------------
// Paging the chunk list
// ---------------------------------------------------------------------------
//
// A file's chunk list ships in a chunkHashes frame, and for a large enough file
// that frame does not fit. @hyperswarm/secret-stream refuses an atomic write over
// 16 MiB-1 and DROPS THE CONNECTION — a 1.25 TB file at tier 3 is ~1.2M entries,
// about 44 MB, so the transfer dies before a single byte of content ships. The
// list is therefore paged, each page but the last carrying more:1.
//
// Each wire entry is 32 B of hash plus a varint length, ~37 B, so 100k entries is
// ~3.7 MB — comfortably under the limit even after Protomux's own send batching.
// The relationship between the page size and the transport limit is asserted in
// test_protocol.cpp rather than left as arithmetic in a comment.
static constexpr size_t MAX_CHUNKS_PER_MSG = 100000;
// @hyperswarm/secret-stream MAX_ATOMIC_WRITE. Not ours to change; it is what the
// page size is chosen against.
static constexpr size_t NOISE_MAX_ATOMIC_WRITE = 16u * 1024u * 1024u - 1u;
// F2: hard ceiling on entries accumulated across a paged stream. Far above any
// real file (that 1.25 TB one is ~1.2M), so it never trips legitimately — it
// bounds a rogue holder that streams more:1 forever. Without it that peer never
// trips the stall watchdog either, because every buffered page counts as progress:
// the two mechanisms would otherwise cover for each other indefinitely.
static constexpr size_t MAX_CHUNK_HASHES_ACCUM = 16u * 1024u * 1024u;
// How long an unanswerable content request waits for serve() to land. Under the
// scheduler's 30s idle timeout on purpose: past that the requester has already
// given up, so holding the request only wastes memory.
static constexpr uint64_t PARKED_REQUEST_TTL_MS = 20000;
// A peer asking for hashes we will never hold must not accumulate. Small, because
// the legitimate case is a handful of assets registered in the same breath.
static constexpr size_t MAX_PARKED_REQUESTS = 64;
// Where a holder's bytes actually live. A LOCATOR, not the bytes: answering a
// request must not cost the file's size in RAM.
struct ContentLocation {
std::string path;
uint64_t size = 0;
};
// An in-progress inbound transfer. The bytes live in a partial on disk; this
// only tracks who is serving it.
struct Download {
std::string contentHash;
std::string destPath;
xfer::Transfer transfer;
sched::ChunkScheduler scheduler;
// Set once the first holder's chunk list has been adopted. Later holders join
// the same fetch rather than restarting it: the list is content-addressed, so
// every honest holder advertises the identical one.
bool started = false;
size_t chunk_count = 0;
};
class Protocol {
public:
// Resolve a content hash to a file this peer can serve. Returning a path
// rather than bytes is the whole point: the overlay's own rule is that
// canonical bytes are real files on disk, and a multi-GB holder must be able
// to answer without reading itself into memory.
using ByteSource = std::function<bool(const std::string &contentHash, ContentLocation &out)>;
Protocol(index::FileIndex *ix, ByteSource src) : index_(ix), source_(std::move(src)) {}
// ---- peers -------------------------------------------------------------
void add_peer(Transport *t) {
if (!t) return;
if (std::find(peers_.begin(), peers_.end(), t) != peers_.end()) return;
peers_.push_back(t);
// A REQUEST IS NOT A ONE-SHOT BROADCAST. request() fans out to whoever is
// connected at that instant; a holder that joins a moment later never saw
// it, stays silent because it was never asked, and the requester's watchdog
// counts down against a peer that would gladly have served it.
//
// That is not hypothetical — it is what layer 12 caught: peers meshed at
// 0.1s, the overlay channel opened later, and the fetch went out to ZERO
// peers and stalled for 30s with two willing holders attached.
//
// Re-sending costs one frame per open download, and a holder that lacks the
// content answers with silence (rule 1), so the newcomer pays nothing for
// content it does not have.
for (auto &kv : downloads_) {
msg::ContentRequest m;
m.contentHash = from_hex32(kv.second->contentHash);
m.from = local_label_;
HO_LOG("peer joined; re-sending in-flight request %s\n",
kv.second->contentHash.substr(0, 12).c_str());
t->send(MsgId::ContentRequest, msg::encode(m));
}
}
void remove_peer(Transport *t) {
for (size_t i = 0; i < parked_.size();)
if (parked_[i].from == t) parked_.erase(parked_.begin() + i); else i++;
peers_.erase(std::remove(peers_.begin(), peers_.end(), t), peers_.end());
serve_auth_.erase(t);
pages_.erase(t);
serve_queue_.erase(t);
serving_now_.erase(t);
blocked_.erase(t);
cancelled_serve_.erase(t);
// Hand its outstanding chunks back so a survivor can take them. Collected
// first because failing a download mutates downloads_.
std::vector<std::string> lost;
for (auto &kv : downloads_) {
kv.second->scheduler.remove_peer(t, now_ms_);
// A download still waiting for its first chunk list has no scheduler
// state to abandon, so the peer list is what decides: with nobody left
// connected, nothing can answer it.
if (kv.second->scheduler.abandoned() || (!kv.second->started && peers_.empty()))
lost.push_back(kv.first);
}
for (const std::string &key : lost) {
auto it = downloads_.find(key);
if (it != downloads_.end()) fail(*it->second, "every holder disconnected");
}
}
size_t peer_count() const { return peers_.size(); }
// Single-peer convenience. Replaces the peer set rather than adding to it, so
// it cannot silently accumulate.
void set_transport(Transport *t) { peers_.clear(); serve_auth_.clear(); add_peer(t); }
// The carrier for this peer drained and can take more. Resumes a serve that
// parked on backpressure. Called by the adapter, which is what learns of it.
void peer_writable(Transport *t) { on_writable(t); }
// The clock. liboverlay never calls now() itself (DESIGN §2), so the host
// drives every timing behaviour through here: the scheduler's no-progress
// watchdog and its hotswap window both hang off it. Not calling tick() does not
// corrupt a transfer; it just means a stalled one is never noticed.
void tick(uint64_t now_ms) {
now_ms_ = now_ms;
// Retry parked requests: serve() may have landed since they arrived.
for (size_t i = 0; i < parked_.size();) {
if (try_answer_content(parked_[i].from, parked_[i].hash) ||
now_ms >= parked_[i].deadline_ms)
parked_.erase(parked_.begin() + i);
else
i++;
}
std::vector<std::pair<std::string, const char *>> dead;
for (auto &kv : downloads_) {
if (!kv.second->scheduler.tick(now_ms)) dead.push_back({ kv.first, "stalled: no progress" });
else if (kv.second->scheduler.abandoned()) dead.push_back({ kv.first, "every holder disconnected" });
}
for (const auto &d : dead) {
auto it = downloads_.find(d.first);
if (it != downloads_.end()) fail(*it->second, d.second);
}
}
void configure_scheduler(const sched::Options &o) { sched_opts_ = o; }
void configure_transfer(const xfer::Options &o) { xfer_opts_ = o; }
// Both default to the constants above; overridable so the paging behaviour can
// be exercised without building a file with 100k chunks in it.
void set_max_chunks_per_message(size_t n) { max_chunks_per_msg_ = n ? n : 1; }
void set_max_chunk_hashes_accum(size_t n) { max_chunk_hashes_accum_ = n; }
// Entries currently held in half-received paged lists. Exposed so the memory a
// peer can make us hold is observable rather than inferred.
size_t buffered_page_entries() const {
size_t n = 0;
for (const auto &kv : pages_) for (const auto &p : kv.second) n += p.second.size();
return n;
}
// Called when a download finishes and verifies. The file is AT dest_path by
// then — handing back a buffer is the thing this layer exists to stop doing.
std::function<void(const std::string &hash, const std::string &dest_path)> on_complete;
// Called when a download ends without the file: a hash that did not match, a
// local write that cannot be retried. Silence otherwise, per rule 1.
std::function<void(const std::string &hash, const std::string &reason)> on_failed;
// ---- requester side ----------------------------------------------------
// Ask every connected peer for a file by content hash, to be assembled at
// `dest_path`. A partial left there by an earlier attempt is resumed.
void request(const std::string &contentHash, const std::string &dest_path) {
const std::string key = content_path(contentHash);
auto d = std::unique_ptr<Download>(new Download());
d->contentHash = contentHash;
d->destPath = dest_path;
// Arm the watchdog NOW, not when a holder answers: a request that nobody
// replies to has to end too, and that is the case where nothing else would
// ever fire.
d->scheduler.configure(sched_opts_);
d->scheduler.arm(now_ms_);
downloads_[key] = std::move(d);
msg::ContentRequest m;
m.contentHash = from_hex32(contentHash);
m.from = local_label_;
const std::vector<uint8_t> frame = msg::encode(m);
HO_LOG("request %s -> fanning out to %zu peer(s)\n", contentHash.substr(0, 12).c_str(), peers_.size());
// Fan out. Holders that do not have it stay silent (rule 1), so asking
// everyone costs one frame per peer and no reply from those without it.
for (Transport *t : peers_) t->send(MsgId::ContentRequest, frame);
}
// ---- inbound -----------------------------------------------------------
void on_message(Transport *from, MsgId id, const std::vector<uint8_t> &frame) {
cenc::State s;
s.buffer = const_cast<uint8_t *>(frame.data());
s.start = 0;
s.end = frame.size();
switch (id) {
case MsgId::ContentRequest: on_content_request(from, msg::decode_content_request(s)); break;
case MsgId::ChunkHashes: on_chunk_hashes_frame(from, msg::decode_chunk_hashes(s)); break;
case MsgId::ChunkNeed: on_chunk_need(from, msg::decode_chunk_need(s)); break;
case MsgId::ChunkData: on_chunk_data(from, msg::decode_chunk_data(s)); break;
case MsgId::ChunkCancel: on_chunk_cancel(from, msg::decode_chunk_cancel(s)); break;
case MsgId::SyncState: break; // handshake noise; nothing to do here
default: break; // unknown/unused slots are ignored, not errors
}
}
// Callers name a content hash; the map is keyed by its synthetic path.
bool has_download(const std::string &h) const { return downloads_.count(content_path(h)) != 0; }
// Stop a fetch. `discard` destroys the partial; otherwise it is kept and the
// next request() for the same hash resumes it — that distinction is the whole
// reason a partial exists, and transfer.h implements both halves already.
//
// ⛔ IT DOES NOT SEND chunkCancel, AND THAT IS NOT AN OVERSIGHT.
//
// The obvious version tells the holders to stop, so they are not pumping
// bytes at a transfer that has stopped reading. It cannot: chunkCancel is a
// HOTSWAP primitive with one-shot semantics — the holder skips that
// (path, index) exactly ONCE and forgets it, because the need it is racing is
// already in flight. Cancelling every outstanding chunk therefore arms one
// landmine per chunk, and the next request for the same content walks into
// all of them: every chunk skipped once, nothing reported by either side, the
// resumed fetch stalled. Measured — that is what the resume assertion below
// caught, and it is why `test_backpressure`'s "the cancelled chunks are
// skipped" is not a rule that can be relaxed to fix it.
//
// The cost of staying silent is bounded and stated: the holders finish the
// serve jobs they already had, and our on_chunk_data drops the bytes because
// the download is gone. For a cancelled transfer that is the tail of one
// file. The alternative is a protocol frame that does not exist, and
// inventing one would stop us talking to the JS peers in production.
bool cancel(const std::string &contentHash, bool discard) {
const std::string path = content_path(contentHash);
auto it = downloads_.find(path);
if (it == downloads_.end()) return false;
Download &d = *it->second;
if (d.started) {
if (discard) d.transfer.cancel();
else d.transfer.pause();
}
drop_pages(path);
downloads_.erase(it);
HO_LOG("cancelled %s (%s)\n", contentHash.substr(0, 12).c_str(),
discard ? "discarded" : "partial kept");
return true;
}
// What a caller can see of a running fetch. False when nothing is running for
// that hash — which is a different answer from "0 of 0" and callers need it.
bool progress(const std::string &contentHash, uint64_t &received_bytes,
size_t &received, size_t &total) const {
auto it = downloads_.find(content_path(contentHash));
if (it == downloads_.end()) return false;
const Download &d = *it->second;
received_bytes = d.transfer.received_bytes();
received = d.transfer.received();
total = d.transfer.total();
return true;
}
// The hashes of every fetch currently running, for a caller that reports on
// all of them rather than polling one it already knows about.
std::vector<std::string> active_downloads() const {
std::vector<std::string> out;
out.reserve(downloads_.size());
for (const auto &kv : downloads_) out.push_back(kv.second->contentHash);
return out;
}
const Download *download(const std::string &h) const {
auto it = downloads_.find(content_path(h));
return it == downloads_.end() ? nullptr : it->second.get();
}
void set_local_label(const std::string &l) { local_label_ = l; }
// Build a file's chunk map without reading the file into memory. Public
// because publishing wants exactly this, and it is the same map the serve path
// caches.
static std::vector<index::ChunkRef> build_chunk_map(const std::string &path, uint64_t size) {
std::vector<index::ChunkRef> refs;
io::File f;
if (!f.open_read(path)) return refs;
const int tier = cdc::select_tier(size);
// Read in blocks at least as large as the tier's maximum chunk, so the
// chunker drains each block instead of re-copying a growing remainder.
const size_t block_size = std::max<size_t>(cdc::tier_params(tier).maxSize, 4u * 1024u * 1024u);
cdc::StreamChunker sc(tier);
std::vector<uint8_t> block(block_size);
auto emit = [&](uint64_t off, const uint8_t *d, size_t len) {
uint8_t h[32];
cdc::hash_chunk(d, len, h);
refs.push_back({ to_hex(h, 32), off, (uint64_t)len });
};
uint64_t off = 0;
while (off < size) {
const size_t want = (size_t)std::min<uint64_t>(block_size, size - off);
if (!f.read_at(off, block.data(), want)) return {}; // changed under us; no map
sc.push(block.data(), want, emit);
off += want;
}
sc.finish(emit);
return refs;
}
private:
// ---- serve side --------------------------------------------------------
// A request we cannot answer YET is not the same as one we cannot answer.
//
// serve() is asynchronous in every real caller: HyperOverlay marshals it onto
// the UDX thread and builds the chunk map there, so a hash is not resolvable
// the instant the game asks for it to be served. A peer that asks inside that
// window got silence and never asked again, and paid the requester's full 30s
// idle watchdog for a file the holder was about to have. Measured: three of
// four concurrent fetches failed with `source resolved=NO`.
//
// The wire stays silent — rule 1 below is right, and a "not found" frame would
// have every peer answer every request. The retry is LOCAL: park it and try
// again on tick. Bounded two ways, because an unanswerable request from a
// peer asking for hashes we will never hold must not accumulate.
void on_content_request(Transport *from, const msg::ContentRequest &m) {
const std::string hash = to_hex(m.contentHash.data(), 32);
if (try_answer_content(from, hash)) return;
for (const Parked &p : parked_)
if (p.from == from && p.hash == hash) return; // already waiting
if (parked_.size() >= MAX_PARKED_REQUESTS) return;
parked_.push_back({ from, hash, now_ms_ + PARKED_REQUEST_TTL_MS });
HO_LOG("contentRequest for %s parked; %zu waiting\n",
hash.substr(0, 12).c_str(), parked_.size());
}
// Answer a content request if we can right now. False = we hold nothing for
// that hash yet, and the caller decides whether to park or drop.
bool try_answer_content(Transport *from, const std::string &hash) {
ContentLocation loc;
const bool resolved = source_ && source_(hash, loc);
HO_LOG("contentRequest for %s arrived; source resolved=%s path=%s\n",
hash.substr(0, 12).c_str(), resolved ? "YES" : "NO",
resolved ? loc.path.c_str() : "-");
// Rule 1: not holding it is SILENCE. No "not found" frame exists in this
// protocol, and inventing one would have every peer answer every request.
if (!resolved) return false;
if (loc.size == 0 && !io::file_size(loc.path, loc.size)) return false;
// The chunk map is content-addressed — identical bytes always chunk
// identically — so a cached one is reused across peers and across
// restarts. Building it costs one streaming pass; doing that per request
// would put a whole-file read in front of the first byte shipped.
const std::vector<index::ChunkRef> *cached =
index_ ? index_->get_chunk_map_by_hash(hash) : nullptr;
std::vector<index::ChunkRef> built;
if (!cached) {
built = build_chunk_map(loc.path, loc.size);
if (built.empty()) return false;
if (index_) index_->put_chunk_map_by_hash(hash, built);
}
const std::vector<index::ChunkRef> &chunks = cached ? *cached : built;
const std::string synth = content_path(hash);
// Serve state is a PATH, keyed by the synthetic path and shared across
// peers — content-addressed, so two peers asking for the same hash want the
// same file. Which peer asked is recorded separately, because that is
// per-peer.
serving_[synth] = loc.path;
serve_auth_[from].insert(synth);
HO_LOG("serving %s: %zu chunks advertised\n", hash.substr(0, 12).c_str(), chunks.size());
send_chunk_hashes(from, synth, (uint8_t)cdc::select_tier(loc.size), chunks);
return true;
}
// Ship a chunk list, paged so no frame can exceed the transport's atomic-write
// limit. A list that fits goes as ONE frame with more:0 — identical on the wire
// to the unpaged format bar a trailing zero byte, which is what keeps this
// compatible with peers that predate paging.
void send_chunk_hashes(Transport *to, const std::string &synth, uint8_t tier,
const std::vector<index::ChunkRef> &chunks) {
const size_t page = max_chunks_per_msg_;
for (size_t off = 0; off == 0 || off < chunks.size(); off += page) {
msg::ChunkHashes out;
out.path = synth;
out.tier = tier;
const size_t end = std::min(chunks.size(), off + page);
for (size_t i = off; i < end; i++) {
msg::ChunkInfo ci;
ci.hash = from_hex32(chunks[i].hash);
ci.length = chunks[i].length;
out.chunks.push_back(ci);
}
out.more = (end < chunks.size()) ? 1 : 0;
emit(to, MsgId::ChunkHashes, msg::encode(out));
if (!out.more) break;
}
}
void on_chunk_need(Transport *from, const msg::ChunkNeed &m) {
// A peer may only pull bytes for a path IT was answered on. Without this a
// peer can skip contentRequest entirely and drive chunkNeed against any
// path some OTHER peer's request registered.
auto sa = serve_auth_.find(from);
if (sa == serve_auth_.end() || sa->second.count(m.path) == 0) return;
auto it = serving_.find(m.path);
if (it == serving_.end()) return;
// The chunk map is keyed by the CONTENT HASH, while the wire carries the
// synthetic path — strip the prefix, as lib/protocol.js:666 does.
if (!index_ || !index_->get_chunk_map_by_hash(content_hash_of(m.path))) return;
ServeJob job;
job.path = m.path;
job.disk_path = it->second;
job.indices = m.indices;
serve_queue_[from].push_back(std::move(job));
// Enqueue, then pump. A second chunkNeed arriving while the first is still
// draining just joins the queue — see drain_serve for why that matters.
drain_serve(from);
}
// The per-peer serve pump. SINGLE-FLIGHT, and that is not an optimisation.
//
// The reference's reason is interleaving: two chunkNeed frames from one peer
// start rival loops that honour backpressure once between them instead of once
// each (lib/protocol.js:679-696, a discipline it lost in a rewrite and put
// back). Ours is worse than that. This loop holds `ServeJob &job` as a
// reference into the deque; a re-entrant call — a carrier that delivers inline,
// which is exactly what an in-memory pair does — pops that job underneath the
// outer frame, and the outer frame then runs on freed memory. Verified: with
// this line removed, test_backpressure's re-entrant case is a use-after-free,
// not a wrong ordering.
//
// PARKS on a full write buffer rather than producing into it. Sibling protocols
// share this stream — the game's own RPC, DHT traffic — and burying them under
// a chunk pump is how a "working" transfer takes the rest of the connection
// down with it.
void drain_serve(Transport *peer) {
if (serving_now_.count(peer)) return; // a pump is already running
if (blocked_.count(peer)) return; // parked until on_writable
serving_now_.insert(peer);
auto &q = serve_queue_[peer];
std::vector<uint8_t> buf;
while (!q.empty()) {
ServeJob &job = q.front();
const std::vector<index::ChunkRef> *refs =
index_ ? index_->get_chunk_map_by_hash(content_hash_of(job.path)) : nullptr;
if (!refs) { q.pop_front(); continue; }
io::File f;
if (!f.open_read(job.disk_path)) { q.pop_front(); continue; }
bool parked = false;
while (job.next < job.indices.size()) {
const uint64_t idx = job.indices[job.next++];
if (idx >= refs->size()) continue; // a peer controls this index
// One-shot: the downloader hotswapped this chunk away, so serving it
// would spend bandwidth on bytes it is already getting elsewhere.
if (cancelled_serve_[peer].erase(job.path + "\x00" + std::to_string(idx))) continue;
const index::ChunkRef &r = (*refs)[(size_t)idx];
buf.resize((size_t)r.length);
if (!f.read_at(r.offset, buf.data(), buf.size())) continue;
msg::ChunkData d;
d.path = job.path;
d.index = idx;
d.data.assign(buf.begin(), buf.end());
if (!emit(peer, MsgId::ChunkData, msg::encode(d))) {
// Buffer full. Stop producing; the job keeps its place so the
// resume picks up at the next index rather than re-sending.
blocked_.insert(peer);
parked = true;
break;
}
}
if (parked) break;
q.pop_front();
}
serving_now_.erase(peer);
}
// The carrier drained. Resume whatever was parked mid-job.
void on_writable(Transport *peer) {
blocked_.erase(peer);
drain_serve(peer);
}
// A downloader reassigned these chunks elsewhere (a hotswap). Skip them if we
// have not served them yet — purely a bandwidth saving, since it races the
// pump and losing that race is harmless.
void on_chunk_cancel(Transport *from, const msg::ChunkCancel &m) {
for (uint64_t i : m.indices)
cancelled_serve_[from].insert(m.path + "\x00" + std::to_string(i));
}
// ---- request side ------------------------------------------------------
// Reassemble a paged chunk list, then dispatch the complete one. A lone
// unpaged frame passes straight through without being copied — the small-file
// norm, and the overwhelming majority of transfers.
void on_chunk_hashes_frame(Transport *from, const msg::ChunkHashes &m) {
// Pages for a path we are not fetching are dropped rather than buffered.
// chunkHashes only ever arrives in answer to our own contentRequest, so
// there is no legitimate case for holding them — and buffering them would
// leave a peer able to grow our memory by naming paths we never asked for.
auto dl = downloads_.find(m.path);
if (dl == downloads_.end()) return;
auto &pages = pages_[from];
auto buffered = pages.find(m.path);
if (buffered == pages.end() && !m.more) {
on_chunk_hashes(from, m.path, m.tier, m.chunks);
return;
}
std::vector<msg::ChunkInfo> &acc =
(buffered == pages.end()) ? pages[m.path] : buffered->second;
acc.insert(acc.end(), m.chunks.begin(), m.chunks.end());
if (acc.size() > max_chunk_hashes_accum_) {
// A holder streaming more:1 forever. Dropping the list fails this
// attempt rather than growing memory without bound; the scheduler
// fails over to another holder.
pages.erase(m.path);
return;
}
if (m.more) {
// Buffering a page is forward progress: a list big enough to need
// paging can take longer to stream than the idle window, and the
// transfer has not started yet, so nothing else would reset it.
dl->second->scheduler.note_progress(now_ms_);
return;
}
std::vector<msg::ChunkInfo> complete;
complete.swap(acc);
pages.erase(m.path);
on_chunk_hashes(from, m.path, m.tier, complete);
}
void on_chunk_hashes(Transport *from, const std::string &path, uint8_t /*tier*/,
const std::vector<msg::ChunkInfo> &chunks) {
auto it = downloads_.find(path);
if (it == downloads_.end()) return;
Download &d = *it->second;
if (d.started) {
// A later holder joins the SAME fetch. The chunk list is content-
// addressed, so every honest holder advertises an identical one; a
// different length means this peer is offering different content under
// our hash, and taking its chunks would just fail verification one at a
// time while wasting its bandwidth and ours.
if (chunks.size() != d.chunk_count) return;
d.scheduler.add_peer(from, now_ms_);
d.scheduler.assign(now_ms_);
return;
}
d.started = true;
d.chunk_count = chunks.size();
xfer::Meta meta;
meta.contentHash = d.contentHash;
uint64_t total = 0;
for (const msg::ChunkInfo &c : chunks) {
xfer::ChunkMeta cm;
std::memcpy(cm.hash, c.hash.data(), 32);
cm.offset = total;
cm.length = c.length;
meta.chunks.push_back(cm);
total += c.length;
}
meta.size = total;
if (!d.transfer.start(d.destPath, meta, xfer_opts_)) {
fail(d, "could not open " + xfer::partial_path_for(d.destPath));
return;
}
// Rule 2: ask only for what is missing. On a resumed partial that is fewer
// than all of them, which is the entire point of keeping one.
const std::vector<uint64_t> want = d.transfer.missing();
if (want.empty()) { complete(d); return; }
d.scheduler.send_need = [this, path](Transport *peer, const std::vector<uint64_t> &idx) {
msg::ChunkNeed need;
need.path = path;
need.indices = idx;
emit(peer, MsgId::ChunkNeed, msg::encode(need));
};
d.scheduler.send_cancel = [this, path](Transport *peer, const std::vector<uint64_t> &idx) {
msg::ChunkCancel cancel;
cancel.path = path;
cancel.indices = idx;
emit(peer, MsgId::ChunkCancel, msg::encode(cancel));
};
d.scheduler.start(want, now_ms_);
d.scheduler.add_peer(from, now_ms_);
d.scheduler.assign(now_ms_);
}
void on_chunk_data(Transport *from, const msg::ChunkData &m) {
auto it = downloads_.find(m.path);
if (it == downloads_.end()) return;
Download &d = *it->second;
if (!d.started) return;
// Bytes are accepted from ANY peer, not just the one they were requested
// from: transfer.write_chunk verifies them against the advertised hash, so
// provenance does not matter. That is what makes multi-source possible at
// all, and it is why a hotswapped peer's late copy is still useful.
const xfer::Transfer::WriteResult r =
d.transfer.write_chunk(m.index, m.data.data(), m.data.size());
if (!r.ok) {
// A bad chunk is one chunk to re-fetch, not a failed transfer — the
// scheduler will offer it to someone else. Only a local write that
// retrying cannot fix ends it here.
if (r.fatal) { fail(d, r.error); return; }
d.scheduler.on_rejected(from, m.index, now_ms_);
return;
}
d.scheduler.on_accepted(from, m.index, now_ms_);
if (d.transfer.complete()) complete(d);
}
// Any half-accumulated pages for this path, on any peer, are stale the moment
// the download settles: a retry of the same hash reuses the same key, and
// appending fresh pages onto an old list would produce a chunk map that never
// existed.
void drop_pages(const std::string &path) {
for (auto &kv : pages_) kv.second.erase(path);
}
void complete(Download &d) {
const xfer::Transfer::FinalizeResult f = d.transfer.finalize();
const std::string hash = d.contentHash;
const std::string dest = d.destPath;
if (!f.ok) {
// A whole-file mismatch means the chunk list itself was wrong or the
// holder lied consistently — the partial is already gone.
fail(d, f.error);
return;
}
drop_pages(content_path(hash));
downloads_.erase(content_path(hash));
if (on_complete) on_complete(hash, dest);
}
void fail(Download &d, const std::string &reason) {
const std::string hash = d.contentHash;
drop_pages(content_path(hash));
downloads_.erase(content_path(hash));
if (on_failed) on_failed(hash, reason);
}
// Returns the carrier's writability, so a caller that is producing in bulk can
// stop. Callers that send a single frame ignore it, which is correct: one frame
// on a full buffer is queued, not lost.
bool emit(Transport *to, MsgId id, const std::vector<uint8_t> &frame) {
return to ? to->send(id, frame) : false;
}
index::FileIndex *index_ = nullptr;
ByteSource source_;
uint64_t now_ms_ = 0;
sched::Options sched_opts_;
xfer::Options xfer_opts_;
// Requests we could not answer yet. Dropped when answered, or at the deadline —
// which is under the requester's idle watchdog, so a request outlives its own
// usefulness rather than lingering.
struct Parked { Transport *from; std::string hash; uint64_t deadline_ms; };
std::vector<Parked> parked_;
size_t max_chunks_per_msg_ = MAX_CHUNKS_PER_MSG;
size_t max_chunk_hashes_accum_ = MAX_CHUNK_HASHES_ACCUM;
// Half-received paged chunk lists, per peer per path.
std::map<Transport *, std::map<std::string, std::vector<msg::ChunkInfo>>> pages_;
std::vector<Transport *> peers_;
std::string local_label_;
// unique_ptr because a Download owns an open file handle and must not be
// copied or moved out from under an in-flight transfer.
std::map<std::string, std::unique_ptr<Download>> downloads_;
// synthetic path -> the file on disk that backs it.
std::map<std::string, std::string> serving_;
// Outstanding serve work, per peer. A job keeps its position so a resume after
// backpressure continues rather than restarting.
struct ServeJob {
std::string path;
std::string disk_path;
std::vector<uint64_t> indices;
size_t next = 0;
};
std::map<Transport *, std::deque<ServeJob>> serve_queue_;
std::set<Transport *> serving_now_; // single-flight guard, per peer
std::set<Transport *> blocked_; // parked on a full write buffer
std::map<Transport *, std::set<std::string>> cancelled_serve_;
// Which synthetic paths each peer was answered on — per peer, because the
// answer is what authorises that peer to pull those bytes.
std::map<Transport *, std::set<std::string>> serve_auth_;
};
} // namespace hyperoverlay
#endif // HYPEROVERLAY_PROTOCOL_H