core: fix quiesce() racing an in-flight write callback (ASan use-after-free)

DownloadTaskState::quiesce() (Engine shutdown / ~Engine, via quiesce_task())
cancelled every live worker's transfer, then immediately cleared `workers` on
the calling thread. transfer.cancel() only *requests* the HttpClient worker
thread stop the transfer -- it does not wait for that to happen. If that
thread was mid write-callback (seg_data -> WriteBuffer::append ->
SparseFile::write_at), clearing the map destroyed the SegWorker (and its
ring buffer) it was still writing through: a heap-use-after-free, caught by
ASan via tools/bench alloc-check, which by design drops its Engine while a
download is still active mid-sample.

Every other exit path (verify/fail/auto_pause/demote, via begin_drain_locked)
already gets this right: cancel, then let each worker remove and flush
itself through seg_finished once HttpClient actually confirms the transfer
stopped, on the correct thread. quiesce() now does the same instead of
tearing the map down itself -- wait on a condition variable, notified from
seg_finished right after it erases, until `workers` is empty.

Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Q3QrF7rCt21bkAjt9BCDFQ

core/src/task/download_task.cpp

@@ -15,6 +15,7 @@
 #include <atomic>
 #include <cctype>
 #include <cerrno>
+#include <condition_variable>
 #include <cstdint>
 #include <cstdlib>
 #include <cstring>
@@ -107,6 +108,9 @@ struct DownloadTaskState : std::enable_shared_from_this<DownloadTaskState> {
     std::mutex mu;
     std::shared_mutex workers_mu;
     std::mutex deferred_mu;
+    // Notified (holding mu) whenever seg_finished removes an entry from `workers`; quiesce()
+    // waits on it instead of clearing the map itself — see quiesce()'s comment.
+    std::condition_variable workers_drained_cv;
 
     EngineState state = EngineState::probing;
     std::optional<ErrorInfo> last_error;
@@ -574,6 +578,7 @@ void DownloadTaskState::seg_finished(std::uint32_t seg_idx, Result<net::Transfer
         w = std::move(it->second);
         workers.erase(it);
     }
+    workers_drained_cv.notify_all();  // quiesce() may be waiting for `workers` to empty out
     if (retired.load()) {  // engine shutting down / already terminal — no more callbacks
         if (w->buf)
             (void)w->buf->flush();
@@ -1184,12 +1189,24 @@ void DownloadTaskState::do_refresh_url(std::string url, std::vector<net::HeaderF
 }
 
 void DownloadTaskState::quiesce() {
-    std::lock_guard lk(mu);
+    std::unique_lock lk(mu);
     retired.store(true);
-    std::unique_lock wl(workers_mu);
-    for (auto &[idx, w] : workers)
-        w->transfer.cancel();
-    workers.clear();
+    {
+        std::shared_lock wl(workers_mu);
+        for (auto &[idx, w] : workers)
+            w->transfer.cancel();
+    }
+    // Do not clear `workers` here: transfer.cancel() only requests the HttpClient worker
+    // thread stop the transfer, asynchronously -- it does not wait for that to happen. A
+    // worker's SegWorker (and its WriteBuffer) may still be in active use by a curl write
+    // callback running on that other thread right now. Clearing the map out from under it
+    // was a real, ASan-caught heap-use-after-free (ring buffer freed here while
+    // SparseFile::write_at() on the HttpClient worker thread was still writing through it).
+    // Every worker removes and flushes itself, safely, via seg_finished once HttpClient
+    // actually confirms the transfer has stopped (same path every other exit uses; see the
+    // `retired` branch there) -- just wait for that to happen for all of them. Bounded by
+    // however long a cancelled curl transfer takes to unwind, not user-controllable.
+    workers_drained_cv.wait(lk, [this] { return workers.empty(); });
 }
 
 EngineState DownloadTaskState::snapshot_state() {