sami/vdm
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

core: drain-aware hostile-mode handling (etag/416/lying-ranges/mismatch)

Browse Source 
Resumes work left mid-session on the M1 hostile-mode matrix. download_task.cpp:

- Replace the old cancel-then-clear teardown (cancel_all_transfers_locked /
  start_assembly_locked) with a single begin_drain_locked()/PendingAction
  mechanism: cancel every live worker, remember what to do (verify / fail /
  auto_pause / demote), and let whichever worker's seg_finished finds the
  worker map empty carry it out. Every sibling still flushes its buffer on
  the way out, so no buffered-but-unflushed tail is lost when a download
  finishes or fails while other segments are still mid-transfer.
- A 200 where 206 was expected (wrong_status) now checks the response's
  ETag/Last-Modified against the probe's: a real mismatch asks the user
  (server_file_changed, "ask, never silently corrupt" -- docs/04 §5); a match
  means the server just stopped honouring Range for this connection, so
  demote to one segment and keep going without a round trip (docs/04 §7).
- 416 mid-download (stale range metadata) now surfaces as a decision instead
  of retrying the same now-invalid range to exhaustion.
- do_decide's abort path surfaces the actual reason a decision was asked
  for (last_error) instead of hardcoding server_file_changed, which was
  mislabeling a 416 abort.

engine_test.cpp adds the four hostile modes where a bug means silent
corruption rather than a visible failure: etag-changes, 416-always,
lies-about-accept-ranges, content-length-mismatch.

Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Q3QrF7rCt21bkAjt9BCDFQ
This commit is contained in:
sami
2026-09-11 13:12:57 +04:00
co-authored by Claude Sonnet 5
parent 0a38867579
commit c99d1d9701
2 changed files with 293 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/task/download_task.cpp
+162 -66
View File
@@ -65,6 +65,17 @@ std::string lower(std::string s) {
} // namespace
// What to do once every worker has drained (see DownloadTaskState::begin_drain_locked).
// A worker that hits one of these outcomes must not tear the task down itself: siblings may
// still be mid-transfer, and cutting them off synchronously — cancel every worker, clear the
// map, proceed — drops their buffered-but-unflushed bytes while segment_completed() already
// counts those bytes as done. That's exactly the class of bug that makes a later resume skip
// real data (seg_data() runs append() without workers_mu, so a sibling's buffer can't be
// flushed safely from here anyway). Instead: cancel the siblings, remember what to do, and
// let each one's own seg_finished (which already flushes on every exit path) run it once the
// worker map is actually empty.
enum class PendingAction { none, verify, fail, auto_pause, demote };
struct SegWorker {
std::uint32_t seg_index = 0;
net::Transfer transfer;
@@ -77,6 +88,7 @@ struct SegWorker {
bool wrong_status = false;
bool range_bad = false;
bool auth_handshake = false; // saw a 401/407 and let libcurl resend with credentials
std::string resp_etag, resp_last_modified; // captured on a wrong_status 200, for demote
std::optional<ErrorInfo> flush_error;
int retries = 0;
@@ -120,12 +132,18 @@ struct DownloadTaskState : std::enable_shared_from_this<DownloadTaskState> {
bool pause_requested = false;
bool cancel_requested = false;
bool assembling = false; // every byte received; draining live workers' buffers to disk
bool discard_on_cancel = false;
bool awaiting_auth = false;
bool awaiting_decision = false;
int max_retries = 10;
// Set by begin_drain_locked while waiting for sibling workers to drain; see
// PendingAction above.
PendingAction pending_action = PendingAction::none;
std::optional<ErrorInfo> pending_error;
bool pending_auth = false;
bool pending_decision = false;
std::atomic<std::int64_t> last_progress_ns{0};
SteadyTime started_at{};
@@ -188,6 +206,7 @@ struct DownloadTaskState : std::enable_shared_from_this<DownloadTaskState> {
void on_probe_result(Result<net::ProbeResult> r);
void finish_probe_locked();
void apply_slot_target(std::uint32_t n);
void fill_slots_locked();
void start_worker_locked(std::uint32_t seg_idx);
void restart_probe(bool with_auth);
@@ -200,8 +219,8 @@ struct DownloadTaskState : std::enable_shared_from_this<DownloadTaskState> {
void begin_verify_locked();
void fail_locked(ErrorInfo e);
void auto_pause_locked(ErrorInfo e, bool auth, bool decision);
void cancel_all_transfers_locked();
void start_assembly_locked();
void demote_to_single_segment_locked();
void begin_drain_locked(PendingAction action, ErrorInfo e, bool auth, bool decision);
void finalize_cancel_locked();
void write_sidecar_locked();
void emit_progress_if_due();
@@ -345,27 +364,37 @@ void DownloadTaskState::finish_probe_locked() {
host.budget().set_want(id, want_slots());
}
// Start workers up to `slot_target`, given whatever the budget currently confirms. Shared
// by apply_slot_target() (the budget's async callback, whenever the computed target
// actually changes) and demote_to_single_segment_locked() -- the demoted target can
// legitimately equal what it was before the rebuild (e.g. a download already down to its
// last segment), in which case SegmentBudget::set_want() no-ops and the async callback
// never fires, so nothing else would ever start the replacement worker.
void DownloadTaskState::fill_slots_locked() {
while (workers.size() < slot_target) {
auto s = seg->assign_slot();
if (!s) {
host.budget().set_want(id, static_cast<std::uint32_t>(workers.size()));
break;
}
if (!host.budget().confirm_slot(id)) {
seg->set_segment_state(*s, segment::SegState::idle);
break;
}
start_worker_locked(*s);
}
if (state == EngineState::connecting && !workers.empty())
transition(EngineState::downloading, std::nullopt);
}
void DownloadTaskState::apply_slot_target(std::uint32_t n) {
{
std::unique_lock lk(mu);
if (retired.load() || is_terminal(state) || pause_requested || cancel_requested ||
awaiting_auth || awaiting_decision || assembling || !seg)
awaiting_auth || awaiting_decision || pending_action != PendingAction::none || !seg)
return;
slot_target = n;
while (workers.size() < slot_target) {
auto s = seg->assign_slot();
if (!s) {
host.budget().set_want(id, static_cast<std::uint32_t>(workers.size()));
break;
}
if (!host.budget().confirm_slot(id)) {
seg->set_segment_state(*s, segment::SegState::idle);
break;
}
start_worker_locked(*s);
}
if (state == EngineState::connecting && !workers.empty())
transition(EngineState::downloading, std::nullopt);
fill_slots_locked();
}
flush_deferred();
}
@@ -464,6 +493,10 @@ net::DataAction DownloadTaskState::seg_head(std::uint32_t seg_idx, const net::Re
// gets a 200 (the source has no Range support).
if (resumable && total_size && *total_size > 0 && h.status == 200) {
w->wrong_status = true;
if (auto v = h.headers.get("ETag"))
w->resp_etag.assign(*v);
if (auto v = h.headers.get("Last-Modified"))
w->resp_last_modified.assign(*v);
return net::DataAction::abort;
}
if (h.status == 416) {
@@ -590,48 +623,87 @@ void DownloadTaskState::seg_finished(std::uint32_t seg_idx, Result<net::Transfer
}
return done();
}
if (assembling) {
// The file is fully received; this worker was cancelled so its buffered tail lands
// on disk. advance() has already counted these bytes; the flush makes them durable.
if (w->buf) {
if (auto f = w->buf->flush(); !f.has_value()) {
release_slot();
fail_locked(std::move(f).error());
return done();
}
}
if (pending_action != PendingAction::none) {
// A sibling already decided the task is finishing (verify / fail / auto-pause /
// demote); this worker's own outcome no longer matters. Drain it like every other
// exit path: flush its buffer so segment_completed() stays true to disk, then hand
// off to whichever worker finds the map empty.
if (w->buf)
(void)w->buf->flush(); // best-effort: we're already tearing down for another
// reason, and the pending action doesn't depend on
// this segment reaching any particular state.
seg->advance(seg_idx, w->base_completed + w->recv);
release_slot();
if (workers.empty())
begin_verify_locked();
if (workers.empty()) {
PendingAction action = std::exchange(pending_action, PendingAction::none);
ErrorInfo e = pending_error.value_or(ErrorInfo(Error::internal, ""));
bool auth = pending_auth, decision = pending_decision;
switch (action) {
case PendingAction::verify:
begin_verify_locked();
break;
case PendingAction::fail:
fail_locked(e);
break;
case PendingAction::auto_pause:
auto_pause_locked(e, auth, decision);
break;
case PendingAction::demote:
demote_to_single_segment_locked();
break;
case PendingAction::none:
break;
}
}
return done();
}
if (w->needs_auth) {
cancel_all_transfers_locked();
release_slot();
auto_pause_locked(ErrorInfo(Error::auth_required, "401/407", w->http_status), true, false);
return done();
}
if (w->wrong_status) {
cancel_all_transfers_locked();
release_slot();
auto_pause_locked(ErrorInfo(Error::server_file_changed, "200 where 206 expected"), false,
true);
// A 200 where 206 was expected is ambiguous: the file really changed (ask, don't
// corrupt — docs/04 §5), or the server just stopped honouring Range for this
// connection while it's still the same file (docs/04 §7: demote to 1 segment and
// continue). ETag/Last-Modified from the 200 itself, compared against what the
// probe recorded, is the only signal that tells them apart. Prefer ETag strictly
// when both sides have one — same rule If-Range itself uses — and only fall back to
// Last-Modified when there's no ETag to compare; a coarse (often second-resolution)
// Last-Modified that happens to match is weak evidence next to a mismatching ETag.
bool same_file;
if (!probe.etag.empty() && !w->resp_etag.empty())
same_file = probe.etag == w->resp_etag;
else if (!probe.last_modified.empty() && !w->resp_last_modified.empty())
same_file = probe.last_modified == w->resp_last_modified;
else
same_file = false; // no validator to compare -> can't prove it, ask
if (same_file) {
demote_to_single_segment_locked();
} else {
auto_pause_locked(ErrorInfo(Error::server_file_changed, "200 where 206 expected"),
false, true);
}
return done();
}
if (w->flush_error) {
ErrorInfo e = *w->flush_error;
release_slot();
if (e.code == Error::disk_full) {
cancel_all_transfers_locked();
auto_pause_locked(e, false, false);
} else {
fail_locked(e);
}
return done();
}
if (w->range_bad)
r = Result<net::TransferStats>(ErrorInfo(Error::range_not_satisfiable, "416"));
if (w->range_bad) {
// 416 mid-download means our range metadata is stale (docs/04 §7): re-probe and
// re-split rather than retrying the same now-invalid range until it exhausts.
release_slot();
auto_pause_locked(ErrorInfo(Error::range_not_satisfiable, "416"), false, true);
return done();
}
if (!r.has_value()) {
ErrorInfo e = std::move(r).error();
@@ -695,17 +767,8 @@ void DownloadTaskState::seg_finished(std::uint32_t seg_idx, Result<net::Transfer
else
release_slot();
if (seg->all_complete()) {
if (workers.empty()) {
begin_verify_locked();
} else {
// Byte counters are satisfied, but other workers are still live and their
// tails may only be in their buffers. Cancel them; each one's seg_finished
// (this thread, once its curl worker has truly stopped) flushes via the
// `assembling` branch, and the last starts verification.
start_assembly_locked();
}
}
if (seg->all_complete())
begin_verify_locked(); // drain-aware: defers if other workers are still live
return done();
}
@@ -713,7 +776,7 @@ void DownloadTaskState::retry_worker(std::uint32_t seg_idx) {
{
std::unique_lock lk(mu);
if (retired.load() || is_terminal(state) || pause_requested || cancel_requested ||
assembling || !seg)
pending_action != PendingAction::none || !seg)
return;
if (workers.count(seg_idx))
return;
@@ -729,6 +792,10 @@ void DownloadTaskState::retry_worker(std::uint32_t seg_idx) {
}
void DownloadTaskState::begin_verify_locked() {
if (!workers.empty()) {
begin_drain_locked(PendingAction::verify, ErrorInfo(Error::internal, ""), false, false);
return;
}
transition(EngineState::assembling, std::nullopt);
transition(EngineState::verifying, std::nullopt);
(void)file->sync();
@@ -778,7 +845,10 @@ void DownloadTaskState::begin_verify_locked() {
}
void DownloadTaskState::fail_locked(ErrorInfo e) {
cancel_all_transfers_locked();
if (!workers.empty()) {
begin_drain_locked(PendingAction::fail, std::move(e), false, false);
return;
}
if (file)
(void)file->close();
if (seg)
@@ -800,6 +870,10 @@ void DownloadTaskState::fail_locked(ErrorInfo e) {
}
void DownloadTaskState::auto_pause_locked(ErrorInfo e, bool auth, bool decision) {
if (!workers.empty()) {
begin_drain_locked(PendingAction::auto_pause, std::move(e), auth, decision);
return;
}
awaiting_auth = auth;
awaiting_decision = decision;
if (file)
@@ -832,26 +906,45 @@ void DownloadTaskState::auto_pause_locked(ErrorInfo e, bool auth, bool decision)
}
}
void DownloadTaskState::cancel_all_transfers_locked() {
std::unique_lock wl(workers_mu);
for (auto &[idx, w] : workers)
w->transfer.cancel();
workers.clear();
}
// Every byte is received but some workers are still live; their buffered tails would be
// lost if we dropped them here (seg_data() runs append() without workers_mu, so we cannot
// safely flush another segment's buffer from under it). Just cancel them and let each
// worker's own seg_finished drain it through the `assembling` branch once its curl worker
// has stopped.
void DownloadTaskState::start_assembly_locked() {
assembling = true;
transition(EngineState::assembling, std::nullopt);
// Cancel every live worker and remember what to do once they've all drained through
// seg_finished's PendingAction branch (see the enum's comment). Never clears `workers`
// itself — each worker removes itself, flushed, when its own transfer actually completes.
void DownloadTaskState::begin_drain_locked(PendingAction action, ErrorInfo e, bool auth,
bool decision) {
pending_action = action;
pending_error = std::move(e);
pending_auth = auth;
pending_decision = decision;
std::shared_lock wl(workers_mu);
for (auto &[idx, w] : workers)
w->transfer.cancel();
}
// The 200-where-206-expected we just saw carried the same ETag/Last-Modified the probe
// recorded: same file, the server (or this connection) just doesn't honour Range. Rebuild
// as a single non-resumable segment covering the whole file and keep going with a plain
// GET. It re-transfers bytes we may already have — there's no way to ask a Range-blind
// server for a suffix — but it never truncates or discards what's on disk, and a source
// that hasn't changed serves identical bytes, so the result is still byte-correct.
void DownloadTaskState::demote_to_single_segment_locked() {
if (!workers.empty()) {
begin_drain_locked(PendingAction::demote, ErrorInfo(Error::internal, ""), false, false);
return;
}
resumable = false;
seg = std::make_unique<segment::Segmenter>(total_size.value_or(0), 1, /*resumable=*/false,
host.config().min_segment_bytes);
transition(EngineState::connecting, std::nullopt);
if (registered) {
// set_want() alone is not enough: if the demoted target happens to equal what it
// was before the rebuild (e.g. this was already the last live segment), it's a
// no-op and the async budget callback never fires. Drive slot assignment directly.
slot_target = want_slots();
host.budget().set_want(id, slot_target);
fill_slots_locked();
}
}
void DownloadTaskState::finalize_cancel_locked() {
if (file)
(void)file->close();
@@ -1025,7 +1118,10 @@ void DownloadTaskState::do_decide(Decision d) {
return;
awaiting_decision = false;
if (d == Decision::abort) {
fail_locked(ErrorInfo(Error::server_file_changed, "user aborted"));
// Surface the reason the decision was actually asked for (range_metadata_stale
// sets last_error to range_not_satisfiable, server_file_changed to itself), not
// a hardcoded label that would misreport a 416 as a changed file.
fail_locked(last_error.value_or(ErrorInfo(Error::server_file_changed, "user aborted")));
} else {
if (d == Decision::restart) {
::unlink(part_path.c_str());
core/tests/task/engine_test.cpp
+131
View File
@@ -62,6 +62,7 @@ struct Recorder {
std::lock_guard lk(mu);
states.push_back(to);
};
c.on_decision_needed = [this](const DecisionRequest &) { decision_calls.fetch_add(1); };
c.on_finished = [this](Result<DownloadOutcome> r) {
if (!fired.exchange(true))
done.set_value(std::move(r));
@@ -326,3 +327,133 @@ VT_TEST(engine_401_then_provide_auth_completes) {
VT_CHECK(rec.auth_calls.load() >= 1);
VT_CHECK_EQ(file_size(td.file("au.bin")), 1u * 1024 * 1024);
}
// --- hostile-mode matrix: the four where a bug is silent corruption, not a visible
// failure (docs/04 §5 "ask, never silently corrupt" / §7's failure-policy table). ---
VT_TEST(engine_etag_changes_asks_instead_of_splicing) {
// A server that revalidates with a different ETag on every response fails an If-Range
// on any retry or resume. That must surface as "ask the user" (server_file_changed),
// never as a silent restart-from-offset-0 spliced onto bytes already on disk.
TestServer srv;
VT_REQUIRE(srv.available());
TmpDir td;
Recorder rec;
Engine eng;
auto h = eng.start(spec_for(srv, "/throttled+etag-changes/file/2M", td.file("ec.bin")),
rec.cbs());
// Get real progress on at least one segment before pausing, so resume's If-Range (only
// sent once a segment has completed > 0) actually fires.
for (int i = 0; i < 300 && h.progress().downloaded < 64u * 1024; ++i)
std::this_thread::sleep_for(10ms);
VT_REQUIRE(h.progress().downloaded >= 64u * 1024);
h.pause();
for (int i = 0; i < 200 && h.state() != EngineState::paused; ++i)
std::this_thread::sleep_for(20ms);
VT_REQUIRE(h.state() == EngineState::paused);
h.resume();
for (int i = 0; i < 300 && rec.decision_calls.load() == 0; ++i)
std::this_thread::sleep_for(20ms);
VT_REQUIRE(rec.decision_calls.load() >= 1);
VT_CHECK_EQ(h.state(), EngineState::paused);
h.decide(Decision::restart);
auto r = rec.wait(90s);
VT_REQUIRE(r.has_value());
VT_CHECK_EQ(file_size(td.file("ec.bin")), 2u * 1024 * 1024);
auto got = hash_file(td.file("ec.bin"), Checksum::Algo::sha256);
VT_CHECK_EQ(got.value(), server_sha(srv, "throttled+etag-changes", "2M"));
}
VT_TEST(engine_416_mid_download_asks_instead_of_exhausting_retries) {
// 416-always 416s every ranged request, including the probe's own -- a live probe
// correctly concludes "not resumable" and a plain-GET download never touches Range
// (that path is the same shape as engine_non_resumable_single_segment). The failure
// mode docs/04 means -- a server that *was* proven resumable dropping Range support
// mid-download -- needs a worker to actually send Range against it, so force the
// resumable, multi-segment assumption directly via probe_hint.
TestServer srv;
VT_REQUIRE(srv.available());
TmpDir td;
Recorder rec;
Engine eng;
net::ProbeResult hint;
hint.total_size = 256u * 1024;
hint.last_modified = "Wed, 01 Jan 2025 00:00:00 GMT";
hint.accept_ranges = true;
hint.resumable = true;
auto s = spec_for(srv, "/416-always/file/256K", td.file("rb.bin"));
s.probe_hint = hint;
s.segments = 2;
auto h = eng.start(std::move(s), rec.cbs());
for (int i = 0; i < 300 && rec.decision_calls.load() == 0; ++i)
std::this_thread::sleep_for(20ms);
VT_REQUIRE(rec.decision_calls.load() >= 1);
VT_CHECK_EQ(h.state(), EngineState::paused);
// 416-always never recovers -- re-probing would just 416 again -- so the only sound
// resolution is to stop, honestly, rather than retry the stale range until exhaustion.
h.decide(Decision::abort);
auto r = rec.wait();
VT_REQUIRE(!r.has_value());
VT_CHECK_EQ(r.error().code, Error::range_not_satisfiable);
VT_CHECK_EQ(::access(td.file("rb.bin").c_str(), F_OK), -1); // never declared complete
}
VT_TEST(engine_lies_about_accept_ranges_demotes_without_asking) {
// Ranges are always ignored (a plain 200, full body) but ETag/Last-Modified are stable
// and honest -- unlike etag-changes, this is provably the *same* file, just a Range-
// blind connection. docs/04 §7: demote to 1 segment and continue, automatically, no
// user round-trip. As with 416-always, a live probe already gets this right up front
// (proven non-resumable), so probe_hint forces the interesting mid-download case.
TestServer srv;
VT_REQUIRE(srv.available());
TmpDir td;
Recorder rec;
Engine eng;
std::string want = server_sha(srv, "lies-about-accept-ranges", "512K");
VT_REQUIRE(!want.empty());
net::ProbeResult hint;
hint.total_size = 512u * 1024;
hint.last_modified = "Wed, 01 Jan 2025 00:00:00 GMT"; // testserver sends this verbatim
hint.accept_ranges = true;
hint.resumable = true;
auto s = spec_for(srv, "/lies-about-accept-ranges/file/512K", td.file("lar.bin"));
s.probe_hint = hint;
s.segments = 4;
auto h = eng.start(std::move(s), rec.cbs());
auto r = rec.wait(60s);
VT_REQUIRE(r.has_value());
VT_CHECK_EQ(rec.decision_calls.load(), 0); // demoted automatically, not asked
VT_CHECK_EQ(file_size(td.file("lar.bin")), 512u * 1024);
auto got = hash_file(td.file("lar.bin"), Checksum::Algo::sha256);
VT_CHECK_EQ(got.value(), want);
}
VT_TEST(engine_content_length_mismatch_fails_honestly) {
// Content-Length promises the true size but the connection always closes short of it.
// There is no recovery (unlike flaky-reset, this never "heals" on a later attempt), so
// the segment's remaining range shrinks every retry until it stalls at zero progress.
// The only correct outcome is a real, visible failure -- never a rename to save_path
// built from a file that is quietly missing bytes.
TestServer srv;
VT_REQUIRE(srv.available());
TmpDir td;
Recorder rec;
Engine eng;
auto s = spec_for(srv, "/content-length-mismatch/file/16K", td.file("clm.bin"));
s.segments = 1;
s.max_retries = 4;
auto h = eng.start(std::move(s), rec.cbs());
auto r = rec.wait(60s);
VT_REQUIRE(!r.has_value());
VT_CHECK_EQ(r.error().code, Error::max_retries_exhausted);
VT_CHECK(rec.saw(EngineState::failed));
VT_CHECK_EQ(::access(td.file("clm.bin").c_str(), F_OK), -1); // never renamed into place
}