diff --git a/core/src/task/download_task.cpp b/core/src/task/download_task.cpp index d67f3be..a613332 100644 --- a/core/src/task/download_task.cpp +++ b/core/src/task/download_task.cpp @@ -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 flush_error; int retries = 0; @@ -120,12 +132,18 @@ struct DownloadTaskState : std::enable_shared_from_this { 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 pending_error; + bool pending_auth = false; + bool pending_decision = false; + std::atomic last_progress_ns{0}; SteadyTime started_at{}; @@ -188,6 +206,7 @@ struct DownloadTaskState : std::enable_shared_from_this { void on_probe_result(Result 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 { 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(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(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, Resultbuf) { - 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(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, Resultall_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(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()); diff --git a/core/tests/task/engine_test.cpp b/core/tests/task/engine_test.cpp index 8e6127a..df687a8 100644 --- a/core/tests/task/engine_test.cpp +++ b/core/tests/task/engine_test.cpp @@ -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 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 +}