#ifndef HYPEROVERLAY_TRANSFER_H
#define HYPEROVERLAY_TRANSFER_H
// One inbound file: bytes land in a partial on disk, are verified as they arrive,
// and the partial is renamed into place only once the whole file hashes to what
// was asked for.
//
// WHAT THIS REPLACES, AND WHY IT IS NOT AN OPTIMISATION. The driver used to
// assemble a download in one std::vector sized to the whole file, and a holder
// materialised the entire file to serve it. That is survivable for a 42 MB asset
// set and wrong for the job — a user-supplied map pack or anything the tier-3
// chunker exists for would spike RAM by the file's whole size, inside a process
// that also has a renderer's budget to respect. The reference never does it: it
// writes into <dest>.overlay-partial at chunk offsets through one persistent
// handle (transfer.js:323, :637). Nothing on the wire can reveal the difference,
// which is exactly why it has to be built deliberately rather than noticed later.
//
// THE FRONTIER, AND WHY THE HASH IS NOT PER CHUNK. The whole-file hash bakes the
// total size into its leaf prefix and then consumes bytes IN ORDER, so it cannot
// absorb chunks as they arrive out of order. Instead it advances over the
// contiguous run of received chunks — once chunk `frontier` is present, feed it
// and move on. By the time the last chunk lands the digest is essentially done,
// so finalize costs no second pass over the file. Out-of-order arrival is the
// normal case the moment more than one peer is serving.
//
// TWO VERIFICATIONS, BOTH NECESSARY. Each chunk is checked against the hash the
// holder advertised, before it reaches the disk. The assembled file is then
// checked against the hash that was ASKED for. The first proves the holder served
// what it advertised; only the second proves that what it advertised is what we
// wanted. The overlay has no serve authorisation — anyone on the topic can
// answer — so neither is optional.
//
// Deliberately NOT here yet: the journal (resume is by re-verification until
// step 3 of DESIGN-transfer-manager.md), and multi-source scheduling.
#include <cstdint>
#include <cstdlib>
#include <cstring>
#include <functional>
#include <map>
#include <set>
#include <string>
#include <vector>
#include "cenc.h"
#include "chunker.h"
#include "fileio.h"
namespace hyperoverlay {
namespace xfer {
// The reference's suffix, kept verbatim. It is deliberately not a plain
// ".partial": host applications sweep that, and a paused overlay download must
// survive their housekeeping (transfer.js:288-291).
static constexpr const char *PARTIAL_SUFFIX = ".overlay-partial";
inline std::string partial_path_for(const std::string &target) { return target + PARTIAL_SUFFIX; }
struct ChunkMeta {
uint8_t hash[32] = {0};
uint64_t offset = 0;
uint64_t length = 0;
};
struct Meta {
uint64_t size = 0;
std::vector<ChunkMeta> chunks;
std::string contentHash; // hex; empty means "no whole-file check possible"
};
// Local write failures split in two. A transient one (device busy, fd pressure,
// an interrupted syscall) may well succeed on retry, so the chunk stays wanted
// and can be re-fetched. Anything else — no space, no permission, read-only
// filesystem — will fail identically for every peer, so retrying across holders
// only wastes their bandwidth to arrive at the same place.
inline bool is_transient_write_error(int e) {
return e == EBUSY || e == EAGAIN || e == EINTR || e == EMFILE || e == ENFILE;
}
// ---------------------------------------------------------------------------
// The journal
// ---------------------------------------------------------------------------
//
// Resuming by re-verification is correct and O(file): every chunk is read back
// and re-hashed to find out what is already there. For a multi-GB asset that is a
// long stall before the first new byte moves. The journal records what the
// transfer already knew — which chunks landed, and the running whole-file digest —
// so a resume is a single small read.
//
// IT LIVES IN APP-PRIVATE STORAGE, NOT NEXT TO THE DOWNLOAD, keyed by a digest of
// the destination path. A journal beside the file would be mistaken for part of
// the download, swept by host housekeeping, or synced somewhere it means nothing.
//
// F1 — NEVER TRUST A JOURNAL YOU CANNOT WHOLE-FILE-VERIFY. This is the rule the
// reference found the hard way (transfer.js:454-461) and the reason the whole
// mechanism is not simply "believe the bitmap":
//
// A stale .overlay-partial for DIFFERENT content can have the same size and
// split into the same number of chunks. Its journal then passes every cheap
// consistency check. If the caller has no content hash to verify against, the
// merged old+new bytes finalize as success and nothing ever notices.
//
// So a journal is only honoured when the caller supplied a content hash AND the
// journal was written for that same hash, size, chunk count and partial path.
// Anything else falls back to re-verification, which re-hashes every chunk and
// therefore cannot be fooled by coincidence.
//
// CRASH ORDERING: fsync the partial BEFORE writing the journal that references
// it. The dangerous direction is a journal claiming chunks whose bytes never
// reached the disk; the reverse (a journal older than the partial) merely
// re-fetches a few chunks that were already there.
//
// There is no checksum on the record, unlike the append-only file index: a
// journal is written whole to a temp file and renamed over the old one, so a torn
// tail cannot exist. A crash mid-write leaves the previous journal intact.
static constexpr const char *JOURNAL_SUFFIX = ".journal";
static constexpr uint32_t JOURNAL_MAGIC = 0x4d4a5633; // "MJV3", as the reference
// Flush cadence is COUNT-based here. The reference also flushes on a 1500ms timer,
// but liboverlay has no clock by design (DESIGN §2) — the owner drives that half
// by calling flush_journal() on its own schedule, and pause() always flushes.
static constexpr size_t JOURNAL_FLUSH_EVERY = 512;
// Journals are named by a digest of the destination path, so they are flat files
// in one directory with no nesting and no path-escaping question.
inline std::string journal_name_for(const std::string &target) {
uint8_t h[32];
cdc::hash_chunk((const uint8_t *)target.data(), target.size(), h);
return to_hex(h, 32) + JOURNAL_SUFFIX;
}
struct Options {
// Cap on just-received chunks held in memory for the hash pump. This is the
// only thing between a fast link and unbounded growth, so it is a refusal,
// not a hint: over the cap, the chunk is simply not stashed and the pump
// reads it back from disk instead.
size_t mem_stash_bytes = 16u * 1024u * 1024u;
// Where journals live. Empty disables journaling entirely: transfers still
// resume, just by re-verification.
std::string journal_dir;
};
class Transfer {
public:
using Options = xfer::Options;
struct WriteResult {
bool ok = false;
std::string error;
int err = 0; // errno, when the failure was local I/O
bool fatal = false; // local I/O that retrying elsewhere cannot fix
};
struct FinalizeResult {
bool ok = false;
std::string error;
bool hash_mismatch = false;
int err_no = 0;
};
~Transfer() { close_fd(); }
// Open (or resume) the transfer. An existing partial of exactly the right size
// is re-verified chunk by chunk and its good chunks are kept; anything else
// starts clean. `on_verify_progress` reports 0..1 across that scan, which can
// be long for a large file and is the reason it is reported at all.
bool start(const std::string &target, const Meta &meta, const Options &opt = Options(),
const std::function<void(double)> &on_verify_progress = nullptr) {
close_fd();
target_ = target;
partial_ = partial_path_for(target);
meta_ = meta;
opt_ = opt;
have_.assign(meta_.chunks.size(), false);
received_count_ = 0;
received_bytes_ = 0;
frontier_ = 0;
stash_.clear();
stash_bytes_ = 0;
readbacks_ = 0;
stash_hits_ = 0;
hashing_ = !meta_.contentHash.empty();
if (hashing_) hasher_.init(meta_.size);
journal_ = opt_.journal_dir.empty() ? std::string()
: opt_.journal_dir + "/" + journal_name_for(target);
since_flush_ = 0;
resumed_from_journal_ = false;
uint64_t existing = 0;
const bool resumable = io::file_size(partial_, existing) && existing == meta_.size;
if (!(resumable ? fd_.open_rw(partial_) : fd_.create_rw(partial_))) return false;
if (!resumable && !fd_.truncate_to(meta_.size)) { close_fd(); return false; }
if (resumable) {
// O(1) if the journal is trustworthy, O(file) if it is not. Never
// "believe it because it is there" — see F1 above.
if (!load_journal()) recover(on_verify_progress);
else resumed_from_journal_ = true;
} else if (!journal_.empty()) {
// A fresh partial invalidates any journal for this destination.
io::unlink_file(journal_);
}
return true;
}
// True when the last start() resumed in O(1) from a journal rather than by
// re-reading the partial. Exposed because "it resumed" and "it resumed without
// re-reading the file" are different claims, and only the second is the point.
bool resumed_from_journal() const { return resumed_from_journal_; }
// Verify one arrived chunk and put it on disk. Verification comes FIRST: bytes
// from a peer are untrusted, and a partial containing unverified bytes is worse
// than no partial, because a later resume would trust it.
WriteResult write_chunk(uint64_t index, const uint8_t *data, size_t len) {
WriteResult r;
if (!fd_.is_open()) { r.error = "transfer closed"; return r; }
if (index >= meta_.chunks.size()) { r.error = "index out of range"; return r; }
const ChunkMeta &c = meta_.chunks[(size_t)index];
if (len != c.length) {
r.error = "length mismatch at " + std::to_string(index);
return r;
}
uint8_t got[32];
cdc::hash_chunk(data, len, got);
if (std::memcmp(got, c.hash, 32) != 0) {
r.error = "hash mismatch at " + std::to_string(index);
return r;
}
if (have_[(size_t)index]) { r.ok = true; return r; } // duplicate; harmless
int e = 0;
if (!fd_.write_at(c.offset, data, len, &e)) {
r.error = "write failed at " + std::to_string(index) + ": " + std::strerror(e);
r.err = e;
r.fatal = !is_transient_write_error(e);
return r;
}
have_[(size_t)index] = true;
received_count_++;
received_bytes_ += len;
// Stash it so the hash pump can consume it without reading it straight back
// off the disk it was just written to. COPIED, never aliased: wire buffers
// are views into shared network slabs decrypted in place, so retaining one
// would pin the slab and bet the digest on its reuse behaviour
// (transfer.js:650-653). Only worth stashing what the pump has not passed.
if (hashing_ && index >= frontier_ && stash_bytes_ + len <= opt_.mem_stash_bytes &&
stash_.find(index) == stash_.end()) {
stash_.emplace(index, std::vector<uint8_t>(data, data + len));
stash_bytes_ += len;
}
advance_hash();
if (++since_flush_ >= JOURNAL_FLUSH_EVERY) flush_journal();
r.ok = true;
return r;
}
// Write the journal now. Called on the count cadence, on pause, and by an owner
// that wants a time-based cadence on top (the half liboverlay's no-clock rule
// leaves to the caller).
bool flush_journal() {
since_flush_ = 0;
// Journaling requires a content hash for the same reason resuming from one
// does: without it there is nothing that could later catch a bad resume.
if (journal_.empty() || !hashing_ || !fd_.is_open()) return false;
// The partial's bytes must be durable BEFORE anything claims they are.
fd_.sync();
std::vector<uint8_t> state;
hasher_.snapshot(state);
const std::vector<uint8_t> bitmap = pack_bitmap();
uint8_t ch[32];
hex_to_32(meta_.contentHash, ch);
const std::vector<uint8_t> rec = cenc::encode_to_vector([&](cenc::State &s, bool pre) {
if (pre) {
cenc::uint32_preencode(s);
cenc::uint_preencode(s, (uint64_t)cdc::StreamingHasher::state_size());
cenc::string_preencode(s, partial_);
cenc::uint_preencode(s, meta_.size);
cenc::uint_preencode(s, (uint64_t)meta_.chunks.size());
cenc::fixed32_preencode(s);
cenc::uint_preencode(s, frontier_);
cenc::uint_preencode(s, hasher_.bytes());
cenc::buffer_preencode(s, state.size());
cenc::buffer_preencode(s, bitmap.size());
} else {
cenc::uint32_encode(s, JOURNAL_MAGIC);
cenc::uint_encode(s, (uint64_t)cdc::StreamingHasher::state_size());
cenc::string_encode(s, partial_);
cenc::uint_encode(s, meta_.size);
cenc::uint_encode(s, (uint64_t)meta_.chunks.size());
cenc::fixed32_encode(s, ch);
cenc::uint_encode(s, frontier_);
cenc::uint_encode(s, hasher_.bytes());
cenc::buffer_encode(s, state.data(), state.size());
cenc::buffer_encode(s, bitmap.data(), bitmap.size());
}
});
// Temp file + rename: the swap is atomic, so a crash mid-write leaves the
// PREVIOUS journal intact rather than a half-written one.
const std::string tmp = journal_ + ".tmp";
FILE *f = std::fopen(tmp.c_str(), "wb");
if (!f) return false;
const bool wrote = std::fwrite(rec.data(), 1, rec.size(), f) == rec.size();
std::fclose(f);
if (!wrote) { io::unlink_file(tmp); return false; }
if (!io::rename_file(tmp, journal_)) { io::unlink_file(tmp); return false; }
return true;
}
// Close the file and put it in place. Refuses on an incomplete transfer, and on
// a whole-file hash that does not match what was asked for — in which case the
// partial is dropped, because resuming from bytes already known to assemble
// wrongly would rebuild the same wrong file.
FinalizeResult finalize() {
FinalizeResult r;
if (received_count_ != meta_.chunks.size()) {
r.error = "incomplete: " + std::to_string(received_count_) + "/" +
std::to_string(meta_.chunks.size());
return r;
}
advance_hash();
if (hashing_) {
const bool whole = frontier_ == meta_.chunks.size();
if (!whole || hasher_.digest() != meta_.contentHash) {
close_fd();
io::unlink_file(partial_);
if (!journal_.empty()) io::unlink_file(journal_);
r.error = "content-hash mismatch";
r.hash_mismatch = true;
return r;
}
}
// The handle must be closed before the rename: an open handle blocks it on
// some platforms, and leaving one open on a file that no longer has that
// name is a leak waiting to be found later.
close_fd();
if (!io::rename_file(partial_, target_)) {
r.error = "rename failed: " + std::string(std::strerror(errno));
r.err_no = errno;
return r;
}
// The transfer is over; its journal describes a partial that no longer
// exists. Leaving it would make the next start() for this destination
// consider a journal for a file that has already been delivered.
if (!journal_.empty()) io::unlink_file(journal_);
r.ok = true;
return r;
}
// Stop, but KEEP the partial so a later start() resumes it. Distinct from
// cancel(), and the distinction is the entire point of having a partial.
//
// Flushes the journal FIRST, while the handle is still open: a paused transfer
// that has to re-verify on resume has thrown away exactly the work the journal
// exists to preserve. The process this runs in is a game, which is killed
// rather than shut down, so this is the common path and not the rare one.
void pause() {
flush_journal();
close_fd();
}
// Stop and destroy the partial. The journal describes a file that will not
// exist, so it goes too — otherwise it is an orphan the next sweep has to
// reason about.
void cancel() {
close_fd();
io::unlink_file(partial_);
if (!journal_.empty()) io::unlink_file(journal_);
}
bool complete() const { return received_count_ == meta_.chunks.size(); }
bool has(uint64_t index) const { return index < have_.size() && have_[(size_t)index]; }
size_t received() const { return received_count_; }
size_t total() const { return meta_.chunks.size(); }
uint64_t received_bytes() const { return received_bytes_; }
const std::string &partial_path() const { return partial_; }
const std::string &target_path() const { return target_; }
// Which chunk indices are still wanted — what a chunkNeed carries. Asking for
// everything works and is exactly the waste resume exists to avoid.
std::vector<uint64_t> missing() const {
std::vector<uint64_t> out;
for (size_t i = 0; i < have_.size(); i++) if (!have_[i]) out.push_back((uint64_t)i);
return out;
}
// Observability for the pump: a transfer that never hits its stash is reading
// every chunk back off disk, which is a real cost and should be visible rather
// than inferred.
size_t readbacks() const { return readbacks_; }
size_t stash_hits() const { return stash_hits_; }
uint64_t hash_frontier() const { return frontier_; }
private:
static void hex_to_32(const std::string &hex, uint8_t out[32]) {
std::memset(out, 0, 32);
for (size_t i = 0; i + 1 < hex.size() && i / 2 < 32; i += 2)
out[i / 2] = (uint8_t)strtoul(hex.substr(i, 2).c_str(), nullptr, 16);
}
std::vector<uint8_t> pack_bitmap() const {
std::vector<uint8_t> bm((have_.size() + 7) / 8, 0);
for (size_t i = 0; i < have_.size(); i++)
if (have_[i]) bm[i >> 3] |= (uint8_t)(1u << (i & 7));
return bm;
}
// Read a journal and adopt it — but only when every reason to distrust it has
// been ruled out. Returns false to mean "re-verify instead", which is always
// safe, never wrong, and merely slow.
bool load_journal() {
if (journal_.empty()) return false;
// F1. With no content hash there is no whole-file verification at finalize
// either, so a journal that is subtly wrong produces a file that is
// silently wrong; re-verification re-hashes every chunk and cannot be
// fooled by a coincidence of size and chunk count.
//
// HONESTY ABOUT COVERAGE: today this is REDUNDANT with the content-hash
// binding below — an empty hash can never equal a 64-char hex string, so
// that check already rejects every such journal, and deleting this line
// alone fails no test. It is kept because the binding is the check a
// well-meaning refactor makes lenient ("don't compare when we have nothing
// to compare against"), and this is the line that still says no afterwards.
// The suite demonstrates the danger by removing BOTH.
if (meta_.contentHash.empty()) return false;
std::vector<uint8_t> raw;
{
io::File f;
if (!f.open_read(journal_)) return false;
uint64_t n = 0;
if (!io::file_size(journal_, n) || n == 0 || n > (64u << 20)) return false;
raw.resize((size_t)n);
if (!f.read_at(0, raw.data(), raw.size())) return false;
}
cenc::State s;
s.buffer = raw.data();
s.start = 0;
s.end = raw.size();
if (cenc::uint32_decode(s) != JOURNAL_MAGIC) return false;
const uint64_t state_bytes = cenc::uint_decode(s);
const std::string stored_partial = cenc::string_decode(s);
const uint64_t size = cenc::uint_decode(s);
const uint64_t total = cenc::uint_decode(s);
uint8_t ch[32];
if (!cenc::fixed32_decode(s, ch)) return false;
const uint64_t frontier = cenc::uint_decode(s);
const uint64_t hbytes = cenc::uint_decode(s);
const cenc::BufView state_v = cenc::buffer_decode(s);
const cenc::BufView bm_v = cenc::buffer_decode(s);
if (!state_v.data || !bm_v.data) return false;
// It must describe THIS transfer. The path/size/count checks are
// belt-and-braces and NOT reachable through the public API — the journal's
// filename is a digest of the destination, and a matching content hash
// implies a matching size and chunk count. They are here for a journal that
// arrives some other way (a shared directory, a restored backup, a manual
// copy), and the suite does not claim to cover them.
if (stored_partial != partial_) return false;
if (size != meta_.size || total != meta_.chunks.size()) return false;
// This one IS load-bearing and is exercised: a decoy that collides on size
// and chunk count gets this far and is rejected only here.
if (to_hex(ch, 32) != meta_.contentHash) return false;
// A hasher state from a different libsodium build cannot be restored, and a
// silently truncated one would poison the digest.
if (state_bytes != cdc::StreamingHasher::state_size()) return false;
if (state_v.len != cdc::StreamingHasher::state_size()) return false;
if (frontier > total) return false;
if (bm_v.len != (meta_.chunks.size() + 7) / 8) return false;
if (!hasher_.restore(state_v.data, state_v.len, hbytes)) return false;
for (size_t i = 0; i < meta_.chunks.size(); i++) {
if (!(bm_v.data[i >> 3] & (1u << (i & 7)))) continue;
have_[i] = true;
received_count_++;
received_bytes_ += meta_.chunks[i].length;
}
frontier_ = frontier;
return true;
}
// Advance the whole-file digest across the contiguous run of received chunks,
// feeding from the stash where possible and from disk otherwise.
void advance_hash() {
if (!hashing_) return;
while (frontier_ < meta_.chunks.size() && have_[(size_t)frontier_]) {
const ChunkMeta &c = meta_.chunks[(size_t)frontier_];
auto it = stash_.find(frontier_);
if (it != stash_.end()) {
hasher_.update(it->second.data(), it->second.size());
stash_bytes_ -= it->second.size();
stash_.erase(it);
stash_hits_++;
} else {
std::vector<uint8_t> buf((size_t)c.length);
if (!fd_.read_at(c.offset, buf.data(), buf.size())) return; // stop; retry later
hasher_.update(buf.data(), buf.size());
readbacks_++;
}
frontier_++;
}
}
// Rebuild what an existing partial already holds, by hashing each chunk where
// it should be. Without a journal this is the only way to know — and it is why
// the journal exists (step 3): this is O(file), not O(1).
void recover(const std::function<void(double)> &progress) {
if (progress) progress(0.0);
std::vector<uint8_t> buf;
for (size_t i = 0; i < meta_.chunks.size(); i++) {
const ChunkMeta &c = meta_.chunks[i];
buf.resize((size_t)c.length);
if (fd_.read_at(c.offset, buf.data(), buf.size())) {
uint8_t got[32];
cdc::hash_chunk(buf.data(), buf.size(), got);
if (std::memcmp(got, c.hash, 32) == 0) {
have_[i] = true;
received_count_++;
received_bytes_ += c.length;
// Feed the contiguous prefix as we go — one pass, not two.
if (hashing_ && i == frontier_) { hasher_.update(buf.data(), buf.size()); frontier_++; }
}
}
if (progress && (i & 63) == 63) progress((double)(i + 1) / (double)meta_.chunks.size());
}
if (progress) progress(1.0);
}
void close_fd() {
fd_.close();
stash_.clear();
stash_bytes_ = 0;
}
std::string target_, partial_, journal_;
size_t since_flush_ = 0;
bool resumed_from_journal_ = false;
Meta meta_;
Options opt_;
io::File fd_;
std::vector<bool> have_;
size_t received_count_ = 0;
uint64_t received_bytes_ = 0;
bool hashing_ = false;
cdc::StreamingHasher hasher_;
uint64_t frontier_ = 0;
std::map<uint64_t, std::vector<uint8_t>> stash_;
size_t stash_bytes_ = 0;
size_t readbacks_ = 0, stash_hits_ = 0;
};
} // namespace xfer
} // namespace hyperoverlay
#endif // HYPEROVERLAY_TRANSFER_H