merge: lane/core

core/docs/m7-baseline.md

@@ -2,9 +2,11 @@
 
 Measured against `docs/04-engine-design.md` §8's targets, via `tools/bench/vdm_bench`
 (see that file's header comment for the exact commands — reproduced below with their
-actual output). `--preset release`, this machine, 2026-09-11. This is a baseline
+actual output). `--preset release`, this machine, 2026-09-11 through 2026-09-12. This is
-record, not a sign-off: two of the three numbers below don't clear the DoD line yet, and
+a baseline record, not a sign-off: one of the three numbers below is loopback-only rather
-that's stated plainly rather than rounded away — see "Open gaps".
+than measured against a real 1 Gbit link, stated plainly rather than rounded away — see
+"Open gaps". The RSS number *did* fail the DoD line in the first pass through this file;
+it's since been root-caused and fixed (`docs/adr/0017`), not just re-measured.
 
 ## Commands and results
 
@@ -25,14 +27,14 @@ the 1-Gbit-saturated cost the target is about). This needs re-running against a
 1 Gbit peer before it can stand as the actual M1/M7 sign-off number.
 
 ```
-$ bin/vdm_bench load --tasks 20 --require-rss-kb 61440
+$ bin/vdm_bench heap-profile --tasks 20 --task-size 4M --top 5
-load: 20 tasks, 0 failed, 17.06s wall
+segment budget: 32 live across tasks, engine reports total=32 active=32 starved=9
-  peak RSS                    69.77 MiB
+heap-profile: peak RSS 45.41 MiB, 20 tasks, 8 segments assumed
-FAIL: peak RSS 71448 KiB > allowed 61440 KiB
 ```
 Default `Config` (`default_segments=8`, `max_active_segments=32`, `default_buffer_bytes=1
-MiB`), default `--task-size 4M`. Correctness holds (0/20 failed); RSS does not clear the
+MiB`), default `--task-size 4M`. **Now clears the 60 MB line** (45.41 MiB), after the real
-60 MB line — see "Open gaps" below.
+fix below — root-caused, not just re-measured. Superseded the original `load` run's ~70
+MiB number quoted in earlier drafts of this file; see `docs/adr/0017`.
 
 ```
 $ bin/vdm_bench alloc-check --size 512M --window-s 2
@@ -48,44 +50,55 @@ fix this bench reported thousands of allocations/sec under any sustained transfe
 
 ## ASan / UBSan / TSan (M1 DoD: "20-task load test... clean")
 
-- `--preset dev` (ASan+UBSan) and `--preset tsan`: the full `core/` test suite (27 ctest
+- `--preset dev` (ASan+UBSan) and `--preset tsan`: the full `core/` test suite (40 ctest
-  cases, including `veloxcore_engine_test`'s hostile-mode suite) and all three
+  cases, including `veloxcore_engine_test`'s hostile-mode suite and `veloxcore_budget_test`)
-  `tools/bench` smoke tests pass clean on both presets.
+  and all three `tools/bench` smoke tests pass clean on both presets.
-- A real bug was caught and fixed getting here: `DownloadTaskState::quiesce()` (engine
+- Two real bugs were caught and fixed getting here:
-  shutdown / `Engine`'s destructor) cleared the `workers` map synchronously right after
+  - `DownloadTaskState::quiesce()` (engine shutdown / `Engine`'s destructor) cleared the
-  issuing an async `transfer.cancel()`, racing the HttpClient worker thread's still-in-flight
+    `workers` map synchronously right after issuing an async `transfer.cancel()`, racing
-  write callback into a heap-use-after-free on the segment's ring buffer — ASan-caught via
+    the HttpClient worker thread's still-in-flight write callback into a heap-use-after-free
-  `alloc-check`, which (by design) drops its `Engine` while a download is still active.
+    on the segment's ring buffer — ASan-caught via `alloc-check`, which (by design) drops
-  Fixed by having `quiesce()` wait for each worker to drain itself through the same
+    its `Engine` while a download is still active. Fixed by having `quiesce()` wait for
-  `seg_finished` path every other exit uses, instead of tearing the map down itself.
+    each worker to drain itself through the same `seg_finished` path every other exit uses,
-- The `tools/bench load` ctest registration runs at reduced concurrency
+    instead of tearing the map down itself.
-  (`--tasks 8 --segments 2`) specifically under sanitizer presets — see
+  - `SegWorker::speed_bps` (the polled-progress fix, see `engine_polled_progress_reports_
-  `tools/bench/CMakeLists.txt`'s comment and `docs/adr/0016`'s postscript for why: at the
+    nonzero_speed`) was written only by a segment's own curl callback and, before this
-  DoD's full 20-tasks × 8-segments shape, `--preset tsan` left an occasional straggler task
+    session, only ever read from that same thread (`emit_progress_if_due`, called from the
-  not completing within a generous per-task budget, with no TSan diagnostic ever
+    same callback) — safe without synchronization. Reading it from `snapshot_progress()`
-  accompanying it. Not proven to be a real engine bug (see the ADR) — filed as a follow-up
+    (any thread calling `DownloadHandle::progress()`) broke that invariant: `workers_mu`'s
-  rather than chased to ground here.
+    shared_lock protects the `workers` map's structure, not an individual `SegWorker`'s
+    mutable fields. TSan-caught. Fixed with `std::atomic<double>` (relaxed: this is an
+    informational EMA, nothing synchronizes real state on it) rather than adding a lock to
+    the write side.
+- The `tools/bench load` ctest registration still runs at reduced concurrency
+  (`--tasks 8 --segments 2`) under sanitizer presets (`tools/bench/CMakeLists.txt`) from
+  when this was written against `docs/adr/0016`'s postscript — see `docs/adr/0017`'s "Open
+  gaps" note: that straggler is now suspected to have been the *same* root cause as the RSS
+  bug, not re-verified at the DoD's full shape under `--preset tsan` in this change.
 
 ## Open gaps
 
-1. **RSS is ~70 MB against a 60 MB target (~10 MB over, ~18%).** `docs/adr/0012` estimated
+1. ~~RSS is ~70 MB against a 60 MB target~~ **Fixed — see `docs/adr/0017`.** Root cause
-   "45–50 MB at the chosen defaults" from segment-buffer arithmetic alone
+   was not, as first guessed here, an `ADR 0012` arithmetic gap or per-queued-handle curl
-   (`max_active_segments=32 * default_buffer_bytes=1 MiB` = 32 MB, plus process/thread-stack
+   overhead: `SegmentBudget::confirm_slot()` only checked a task's *own* target against
-   fixed cost). A minimal single-tiny-task run here measured that fixed cost at ~14.7 MB,
+   its own held count, never the engine-wide `active_` sum, so it could (and under real
-   which lines up with the ADR's estimate (32 + 15 ≈ 47 MB) — but the real 20-task number is
+   20-task/8-segment contention, reliably did) admit segments well past
-   ~20 MB higher than that. Not root-caused in this change: a plausible next step is
+   `max_active_segments` — `heap-profile` caught it directly: `budget.active` reading
-   checking whether `net::HttpClient` holds a live `curl_easy` handle (and its own internal
+   56–86 against a `total` of 32. Fixed at the budget level (the one place that can
-   buffers) per *queued* segment, not just per *active* one — 20 tasks × 8 segments = 160
+   actually enforce the invariant); `docs/adr/0012`'s own arithmetic was fine all along.
-   queued handles even though only 32 run concurrently, which would explain a gap this
-   ADR's arithmetic (32 *active* buffers) doesn't account for.
 2. **Throughput/CPU numbers are loopback-only.** No 1 Gbit link was available to test
    against; re-run `throughput --size 5G --require-mbps 940 --max-cpu-pct 8` against a real
    one before treating this as signed off.
-3. **`docs/adr/0016`**: `rate::RateLimiter`'s global-limit path has no fairness ordering
+3. **`docs/adr/0016`**: `rate::RateLimiter`'s global-limit path (byte-rate pacing, a
-   under heavy segment contention (a shared `TokenBucket`'s peek/commit race can starve a
+   different subsystem from the segment-admission bug in `docs/adr/0017`) has no fairness
-   waiter indefinitely) — a real gap for the "global bandwidth cap with many concurrent
+   ordering under heavy segment contention (a shared `TokenBucket`'s peek/commit race can
-   downloads" scenario, filed there rather than fixed in this change.
+   starve a waiter indefinitely) — a real, separate, still-open gap for the "global
-4. **The TSan-only load-test straggler** noted above (`docs/adr/0016`'s postscript) —
+   bandwidth cap with many concurrent downloads" scenario.
-   not root-caused; needs reproducing outside a shared/virtualized sandbox to tell "TSan is
+4. **The TSan-only load-test straggler** noted in `docs/adr/0016`'s postscript, found
-   just slow here" apart from a real timing-sensitive bug (a plausible candidate named in
+   before `docs/adr/0017`'s fix landed: plausibly the *same* root cause (a segment denied
-   the ADR: `CURLOPT_LOW_SPEED_TIME` false-tripping under TSan's slowdown).
+   admission with nothing to wake it, worse under TSan's slowdown widening the window a
+   deferred yield can sit in) rather than the `CURLOPT_LOW_SPEED_TIME` guess that ADR
+   originally offered — not reverified at the DoD's full 20-task/8-segment shape under
+   `--preset tsan` in this change (the sanitizer-preset smoke registration still runs at
+   reduced concurrency; see `tools/bench/CMakeLists.txt`). Worth re-running before treating
+   it as closed.

core/include/vdm/segment/budget.hpp

@@ -100,6 +100,10 @@ class SegmentBudget {
         std::uint32_t target = 0;  // last published
         SlotTargetFn on_target;
         std::optional<SteadyTime> starved_since;
+        // confirm_slot() was denied by the engine-wide cap (not by this task's own
+        // target) and hasn't been retried since. Cleared on the task's next successful
+        // confirm_slot(), however that retry was triggered. See release_slot()'s comment.
+        bool waiting_for_slot = false;
     };
 
     // A unit of deferred work: callbacks are copied out here so the public entry points
@@ -109,7 +113,12 @@ class SegmentBudget {
         std::optional<std::pair<std::function<void(EngineBudget)>, EngineBudget>> notify_now;
     };
 
+    [[nodiscard]] std::vector<TaskId> priority_order_locked() const;
     Plan reallocate_locked();
+    // Appends a retry hint for the highest-priority task with waiting_for_slot set (other
+    // than `exclude`, the task whose release just freed this slot -- see release_slot()'s
+    // comment) to `plan`, if one exists.
+    void wake_one_waiter_locked(Plan &plan, TaskId exclude) const;
     static void run(Plan &p);
     [[nodiscard]] std::uint32_t effective_cap_locked(const Task &t) const;
     [[nodiscard]] EngineBudget snapshot_locked() const;

core/src/segment/budget.cpp

@@ -37,12 +37,10 @@ SegmentBudget::EngineBudget SegmentBudget::snapshot_locked() const {
     return EngineBudget{max_active_, active_, starved};
 }
 
-// The two-pass fairness allocation. Recomputes every task's target from scratch (so a
+// DAEMON's list first, then any registered task not in it (defensive; "a running task
-// live cap cut naturally produces target < held -> yield), diffs against the last
+// absent from the list sorts last"). Shared by reallocate_locked() and
-// published target, and collects the callbacks to fire once mu_ is released.
+// wake_one_waiter_locked(), which need the same priority ordering.
-SegmentBudget::Plan SegmentBudget::reallocate_locked() {
+std::vector<TaskId> SegmentBudget::priority_order_locked() const {
-    // Priority order: DAEMON's list first, then any registered task not in it (defensive;
-    // "a running task absent from the list sorts last").
     std::vector<TaskId> order;
     order.reserve(tasks_.size());
     for (TaskId id : order_)
@@ -51,6 +49,14 @@ SegmentBudget::Plan SegmentBudget::reallocate_locked() {
     for (const auto &[id, _] : tasks_)
         if (std::find(order.begin(), order.end(), id) == order.end())
             order.push_back(id);
+    return order;
+}
+
+// The two-pass fairness allocation. Recomputes every task's target from scratch (so a
+// live cap cut naturally produces target < held -> yield), diffs against the last
+// published target, and collects the callbacks to fire once mu_ is released.
+SegmentBudget::Plan SegmentBudget::reallocate_locked() {
+    std::vector<TaskId> order = priority_order_locked();
 
     std::unordered_map<TaskId, std::uint32_t> target;
     target.reserve(order.size());
@@ -155,6 +161,7 @@ void SegmentBudget::deregister_task(TaskId id) {
         active_ -= it->second.held;
         tasks_.erase(it);
         plan = reallocate_locked();
+        wake_one_waiter_locked(plan, id);  // a departing task frees real slots too
     }
     run(plan);
 }
@@ -182,6 +189,19 @@ bool SegmentBudget::confirm_slot(TaskId id) {
     Task &t = it->second;
     if (t.held >= t.target)
         return false;  // target was cut in the race
+    // reallocate_locked()'s pool math bounds sum(target) <= max_active_ *as computed*, but
+    // that doesn't bound sum(held): a task can be legitimately over its own just-lowered
+    // target for a while (yield deferred to a segment boundary, ADR 0011 A1), and another
+    // task's target can correctly rise to claim that capacity before the first task has
+    // physically released it. active_ is the one number that's always true regardless of
+    // any task's target bookkeeping, so it's the backstop. Denials here are remembered
+    // (waiting_for_slot) rather than left for the caller to somehow ask again at the right
+    // moment -- see release_slot()'s wake_one_waiter_locked() call.
+    if (active_ >= max_active_) {
+        t.waiting_for_slot = true;
+        return false;
+    }
+    t.waiting_for_slot = false;
     ++t.held;
     ++active_;
     if (snapshot_locked() != last_notified_) {
@@ -191,6 +211,25 @@ bool SegmentBudget::confirm_slot(TaskId id) {
     return true;
 }
 
+// Appends a retry hint for the highest-priority task with waiting_for_slot set (other
+// than `exclude`) to `plan`. Only ever wakes a task confirm_slot() actually turned away --
+// not just anyone below its target, which would also fire for tasks that are fairly,
+// correctly not entitled to more right now (see reallocate_locked()'s own target math).
+void SegmentBudget::wake_one_waiter_locked(Plan &plan, TaskId exclude) const {
+    for (TaskId id : priority_order_locked()) {
+        if (id == exclude)
+            continue;
+        auto it = tasks_.find(id);
+        if (it == tasks_.end())
+            continue;
+        const Task &t = it->second;
+        if (t.waiting_for_slot && t.on_target) {
+            plan.targets.emplace_back(t.on_target, t.target);
+            return;  // exactly one freed slot, exactly one retry hint
+        }
+    }
+}
+
 void SegmentBudget::release_slot(TaskId id) {
     Plan plan;
     {
@@ -201,6 +240,11 @@ void SegmentBudget::release_slot(TaskId id) {
         --it->second.held;
         --active_;
         plan = reallocate_locked();
+        // reallocate_locked() only fires a callback for a task whose *target* changed.
+        // The task this freed slot is actually owed to (see confirm_slot()) may have a
+        // target that was already correct and hasn't moved -- nothing else will ever ask
+        // it to retry, so the budget has to remember and hand this slot to it directly.
+        wake_one_waiter_locked(plan, id);
     }
     run(plan);
 }

core/src/task/download_task.cpp

@@ -96,7 +96,17 @@ struct SegWorker {
 
     SteadyTime sample_at{};
     std::uint64_t sample_bytes = 0;
-    double speed_bps = 0;
+    // Written only by this segment's own curl callback (seg_data, sequential -- no lock
+    // held across the update, by design: the transfer hot path takes no lock it doesn't
+    // need). snapshot_progress() reads it from whatever thread calls
+    // DownloadHandle::progress() (DAEMON's polling, or anyone else's), which workers_mu's
+    // shared_lock does NOT cover -- that lock only protects the `workers` map's own
+    // structure, not an individual SegWorker's mutable fields. atomic<double> (relaxed:
+    // this is an approximate, informational EMA, not something anything synchronizes
+    // real state on) keeps that read-from-any-thread safe without adding a lock to the
+    // write side. Found by TSan the first time anything actually read this cross-thread
+    // (engine_polled_progress_reports_nonzero_speed, added alongside the speed_bps fix).
+    std::atomic<double> speed_bps{0};
 };
 
 struct DownloadTaskState : std::enable_shared_from_this<DownloadTaskState> {
@@ -211,6 +221,11 @@ struct DownloadTaskState : std::enable_shared_from_this<DownloadTaskState> {
     void finish_probe_locked();
     void apply_slot_target(std::uint32_t n);
     void fill_slots_locked();
+    // Confirms a budget slot and starts a worker for `seg_idx` if nothing already covers
+    // it. false on either failure (already running, or budget denied) -- the caller's own
+    // fallback (fill_slots_locked's assign_slot() loop, retry_worker()'s early return)
+    // stays the same either way. Caller holds mu.
+    bool try_start_segment_locked(std::uint32_t seg_idx);
     void start_worker_locked(std::uint32_t seg_idx);
     void restart_probe(bool with_auth);
 
@@ -368,6 +383,15 @@ void DownloadTaskState::finish_probe_locked() {
     host.budget().set_want(id, want_slots());
 }
 
+bool DownloadTaskState::try_start_segment_locked(std::uint32_t seg_idx) {
+    if (workers.count(seg_idx))
+        return false;
+    if (!host.budget().confirm_slot(id))
+        return false;
+    start_worker_locked(seg_idx);
+    return true;
+}
+
 // 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
@@ -375,6 +399,23 @@ void DownloadTaskState::finish_probe_locked() {
 // 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() {
+    // A segment mid-backoff (SegState::stalled -- seg_finished's retry path: released its
+    // slot, scheduled a retry_worker() timer, and gave up quietly if confirm_slot() denied
+    // it then) has no live worker and never surfaces through assign_slot() -- it stays
+    // assigned to whichever segment iteration created it, just not running. It's not
+    // "fresh work" the loop below would ever find on its own. Every path that can free
+    // budget capacity ends up here (apply_slot_target(), driven by SegmentBudget's
+    // target-changed callback *and* its wait-list wakeup for a task whose target didn't
+    // move -- see confirm_slot()/release_slot() in budget.cpp), so this is the one place
+    // that needs to give a stalled segment another try, not every caller of
+    // retry_worker(). Bounded by slot_target like the assign_slot() loop below; a segment
+    // this doesn't get to keeps its own scheduled retry_worker() timer as a second chance.
+    for (auto &v : seg->snapshot()) {
+        if (workers.size() >= slot_target)
+            break;
+        if (v.state == segment::SegState::stalled)
+            try_start_segment_locked(v.index);
+    }
     while (workers.size() < slot_target) {
         auto s = seg->assign_slot();
         if (!s) {
@@ -387,7 +428,11 @@ void DownloadTaskState::fill_slots_locked() {
         }
         start_worker_locked(*s);
     }
-    if (state == EngineState::connecting && !workers.empty())
+    // Mirrors retry_worker()'s own transition: a stalled-segment restart above can be the
+    // thing that takes a retry_wait task back to actually transferring, same as connecting
+    // does for a task starting up.
+    if ((state == EngineState::connecting || state == EngineState::retry_wait) &&
+        !workers.empty())
         transition(EngineState::downloading, std::nullopt);
 }
 
@@ -547,7 +592,9 @@ net::DataAction DownloadTaskState::seg_data(std::uint32_t seg_idx, ConstByteSpan
     auto dt = std::chrono::duration<double>(now - w->sample_at).count();
     if (dt >= 0.5) {
         double inst = static_cast<double>(w->recv - w->sample_bytes) / dt;
-        w->speed_bps = w->speed_bps == 0 ? inst : 0.7 * w->speed_bps + 0.3 * inst;
+        double prev = w->speed_bps.load(std::memory_order_relaxed);
+        w->speed_bps.store(prev == 0 ? inst : 0.7 * prev + 0.3 * inst,
+                           std::memory_order_relaxed);
         w->sample_at = now;
         w->sample_bytes = w->recv;
     }
@@ -1013,10 +1060,11 @@ void DownloadTaskState::emit_progress_if_due() {
         std::shared_lock lk(workers_mu);
         double agg = 0;
         for (auto &[idx, w] : workers) {
-            agg += w->speed_bps;
+            double speed = w->speed_bps.load(std::memory_order_relaxed);
+            agg += speed;
             SegmentProgress sp;
             sp.index = idx;
-            sp.speed_bps = static_cast<std::uint64_t>(w->speed_bps);
+            sp.speed_bps = static_cast<std::uint64_t>(speed);
             p.segments.push_back(sp);
         }
         p.speed_bps = static_cast<std::uint64_t>(agg);
@@ -1217,6 +1265,26 @@ EngineState DownloadTaskState::snapshot_state() {
 Progress DownloadTaskState::snapshot_progress() {
     Progress p;
     std::lock_guard lk(mu);
+    // Per-segment instantaneous speed lives on the live SegWorker (seg_data's 0.5s-sampled
+    // EMA, see the `speed_bps` update below) -- a segment with no live worker (idle,
+    // paused, complete, failed) has no speed to report and stays at SegmentProgress's
+    // default 0. Read every live worker's speed up front so the seg->snapshot() loop below
+    // (which covers *every* segment, not just live ones -- unlike emit_progress_if_due's
+    // push-callback version, which only ever reports the segments it currently has
+    // workers for) can look each one up by index.
+    double agg_speed = 0;
+    std::unordered_map<std::uint32_t, double> worker_speed;
+    {
+        std::shared_lock wl(workers_mu);
+        worker_speed.reserve(workers.size());
+        for (auto &[idx, w] : workers) {
+            double speed = w->speed_bps.load(std::memory_order_relaxed);
+            worker_speed.emplace(idx, speed);
+            agg_speed += speed;
+        }
+        p.effective_segments = static_cast<std::uint32_t>(workers.size());
+    }
+    p.speed_bps = static_cast<std::uint64_t>(agg_speed);
     if (seg) {
         p.downloaded = seg->downloaded();
         for (auto &v : seg->snapshot()) {
@@ -1226,15 +1294,15 @@ Progress DownloadTaskState::snapshot_progress() {
             sp.end = v.end;
             sp.completed = v.completed;
             sp.state = v.state;
+            if (auto it = worker_speed.find(v.index); it != worker_speed.end())
+                sp.speed_bps = static_cast<std::uint64_t>(it->second);
             p.segments.push_back(sp);
         }
     }
     p.total = total_size;
-    {
-        std::shared_lock wl(workers_mu);
-        p.effective_segments = static_cast<std::uint32_t>(workers.size());
-    }
     p.effective_buffer_bytes = effective_buffer;
+    if (p.speed_bps > 0 && total_size && *total_size > p.downloaded)
+        p.eta_seconds = static_cast<std::uint32_t>((*total_size - p.downloaded) / p.speed_bps);
     return p;
 }
 

core/tests/segment/budget_test.cpp

@@ -2,7 +2,10 @@
 
 #include <atomic>
 #include <chrono>
+#include <condition_variable>
+#include <functional>
 #include <mutex>
+#include <optional>
 #include <thread>
 #include <vector>
 
@@ -20,29 +23,153 @@ TaskId tid(std::uint64_t v) {
 
 // A test task that reacts to slot targets the way stage 8's download_task will: start
 // workers up to the target, release them when the target drops. Purely bookkeeping.
+//
+// Production's real equivalent (register_task()'s on_target lambda, download_task.cpp)
+// never calls back in synchronously -- it posts through host.schedule() (engine.cpp), so
+// apply_slot_target() never runs on the same call stack as whatever budget call triggered
+// it. wake_one_waiter_locked()'s wakeups, though, mean on_target() here *can* now be
+// reentered on the same thread (e.g. this task's own release_slot() call, deep in the grow
+// loop below, can cascade into waking a *different* task whose own release, in turn,
+// cascades back to this one) -- a plain std::mutex would self-deadlock on that, the way
+// FakeTask::mu almost did until this was written. A recursive_mutex plus coalescing the
+// reentrant call's target into `pending` (processed by the outer call's loop once its
+// current pass finishes) keeps this a faithful, deterministic stand-in without actually
+// needing real threads: no callback ever runs nested inside a still-running instance of
+// itself, and a burst of reentrant retargeting converges to the latest value instead of
+// each one doing its own redundant grow/shrink pass.
 struct FakeTask {
     SegmentBudget *budget = nullptr;
     TaskId id{};
-    std::mutex mu;
+    std::recursive_mutex mu;
     std::uint32_t workers = 0;
     std::uint32_t target = 0;
+    bool running = false;
+    std::optional<std::uint32_t> pending;
 
     FakeTask() = default;
     FakeTask(SegmentBudget *b, TaskId i) : budget(b), id(i) {}
 
     void on_target(std::uint32_t t) {
         std::lock_guard lk(mu);
-        target = t;
+        if (running) {
-        while (workers < target) {
+            pending = t;  // reentrant call on this thread -- the running instance picks it up
-            if (!budget->confirm_slot(id))
+            return;
+        }
+        running = true;
+        std::uint32_t want = t;
+        for (;;) {
+            target = want;
+            while (workers < target) {
+                if (!budget->confirm_slot(id))
+                    break;
+                ++workers;
+            }
+            // over target -> yield the excess immediately (a real task waits for a
+            // boundary)
+            while (workers > target) {
+                budget->release_slot(id);
+                --workers;
+            }
+            if (!pending)
                 break;
-            ++workers;
+            want = *pending;
+            pending.reset();
         }
-        // over target -> yield the excess immediately (a real task waits for a boundary)
+        running = false;
-        while (workers > target) {
+    }
-            budget->release_slot(id);
+    std::uint32_t held() {
-            --workers;
+        std::lock_guard lk(mu);
+        return workers;
+    }
+};
+
+// Mirrors Engine's real timer thread (engine.cpp: timer_loop) and how register_task()'s
+// on_target lambda actually reaches a task (download_task.cpp: host.schedule(), never a
+// direct call). One dedicated thread drains queued closures one at a time -- the
+// serialization that makes production immune to the cross-task deadlock risk FakeTask's
+// synchronous, whatever-thread-triggered-it delivery has under heavy *concurrent* driving
+// (budget_concurrent_confirm_release_stays_consistent, below, is the one test that
+// actually exercises this: multiple threads calling set_want() for *different* tasks at
+// once, each able to cascade into a wake for another task's callback -- two such cascades
+// landing on two different FakeTask mutexes in opposite orders on two different threads is
+// a real AB-BA deadlock a recursive_mutex alone doesn't prevent, since that only guards a
+// single thread against re-entering itself). Every other test in this file drives the
+// budget from one thread at a time, where that risk can't arise, so plain FakeTask is
+// still the simpler, sufficient double there.
+class TestTimer {
+  public:
+    TestTimer() {
+        worker_ = std::jthread([this](std::stop_token st) { run(st); });
+    }
+    ~TestTimer() {
+        worker_.request_stop();
+        cv_.notify_all();
+    }
+    void post(std::function<void()> fn) {
+        {
+            std::lock_guard lk(mu_);
+            queue_.push_back(std::move(fn));
         }
+        cv_.notify_one();
+    }
+    // Blocks until the queue is empty and nothing is mid-run -- what a test needs before
+    // asserting on state this timer's closures mutate.
+    void drain() {
+        std::unique_lock lk(mu_);
+        cv_done_.wait(lk, [&] { return queue_.empty() && !running_; });
+    }
+
+  private:
+    void run(std::stop_token st) {
+        std::unique_lock lk(mu_);
+        while (true) {
+            cv_.wait(lk, st, [&] { return !queue_.empty(); });
+            if (st.stop_requested())
+                return;
+            auto fn = std::move(queue_.front());
+            queue_.erase(queue_.begin());
+            running_ = true;
+            lk.unlock();
+            fn();
+            lk.lock();
+            running_ = false;
+            if (queue_.empty())
+                cv_done_.notify_all();
+        }
+    }
+    std::mutex mu_;
+    std::condition_variable_any cv_;  // _any: wait() below takes a stop_token predicate
+    std::condition_variable cv_done_;
+    std::vector<std::function<void()>> queue_;
+    bool running_ = false;
+    std::jthread worker_;
+};
+
+// Same grow/shrink logic as FakeTask, but on_target() only ever posts through a TestTimer
+// instead of running synchronously -- see the comment above TestTimer for why that's the
+// faithful model under concurrent driving.
+struct AsyncFakeTask {
+    SegmentBudget *budget = nullptr;
+    TestTimer *timer = nullptr;
+    TaskId id{};
+    std::mutex mu;  // only the timer thread ever touches workers/target -- plain suffices
+    std::uint32_t workers = 0;
+    std::uint32_t target = 0;
+
+    void on_target(std::uint32_t t) {
+        timer->post([this, t] {
+            std::lock_guard lk(mu);
+            target = t;
+            while (workers < target) {
+                if (!budget->confirm_slot(id))
+                    break;
+                ++workers;
+            }
+            while (workers > target) {
+                budget->release_slot(id);
+                --workers;
+            }
+        });
     }
     std::uint32_t held() {
         std::lock_guard lk(mu);
@@ -235,12 +362,14 @@ VT_TEST(budget_on_changed_fires_on_starved_edge) {
 VT_TEST(budget_concurrent_confirm_release_stays_consistent) {
     SegmentBudget b({.max_active_segments = 16});
     constexpr int kTasks = 6;
-    std::vector<std::unique_ptr<FakeTask>> ts;
+    TestTimer timer;
+    std::vector<std::unique_ptr<AsyncFakeTask>> ts;
     for (int i = 0; i < kTasks; ++i) {
-        ts.push_back(std::make_unique<FakeTask>());
+        ts.push_back(std::make_unique<AsyncFakeTask>());
         ts.back()->budget = &b;
+        ts.back()->timer = &timer;
         ts.back()->id = tid(i + 1);
-        FakeTask *ft = ts.back().get();
+        AsyncFakeTask *ft = ts.back().get();
         b.register_task(ft->id, {.host = "h", .per_task_cap = 6, .resumable = true},
                         [ft](std::uint32_t n) { ft->on_target(n); });
     }
@@ -254,6 +383,7 @@ VT_TEST(budget_concurrent_confirm_release_stays_consistent) {
     drivers.clear();  // join
     for (auto &ft : ts)
         b.set_want(ft->id, 0);
+    timer.drain();  // let every queued on_target actually run before asserting
 
     // With everyone wanting nothing, the budget must be fully released.
     VT_CHECK_EQ(b.budget().active, 0u);
@@ -262,3 +392,123 @@ VT_TEST(budget_concurrent_confirm_release_stays_consistent) {
         sum += b.segments_active(ft->id);
     VT_CHECK_EQ(sum, 0u);
 }
+
+// --- wait-list wakeup: tools/bench heap-profile / load found a real task time out
+// waiting on a slot its own target already said it should have (core/docs/m7-baseline.md,
+// docs/adr/0012). Root cause: confirm_slot()'s engine-wide cap check (needed so active_
+// never exceeds max_active_segments -- a real over-admission bug, not just this liveness
+// gap) can deny a task whose target is already correct, when a *different* task is
+// legitimately still holding more than its own just-lowered target (yield is deferred to
+// a segment boundary, ADR 0011 A1). Nothing in the plain target-changed callback
+// mechanism ever revisits a task whose target didn't change -- it was already right. ---
+
+namespace {
+
+// Unlike FakeTask above, on_target() here only enqueues -- it never calls back into the
+// budget synchronously. This matches production exactly: register_task()'s on_target
+// lambda (download_task.cpp) posts through host.schedule() (engine.cpp), so
+// apply_slot_target() never runs on the same call stack as whatever budget call triggered
+// it. The test drives delivery explicitly (deliver_one()) instead of a background thread
+// so the race this test exists to force -- confirm_slot() denied before the task that's
+// over its target has processed its own shrink -- is deterministic, not a timing gamble.
+struct QueuedTask {
+    SegmentBudget *budget = nullptr;
+    TaskId id{};
+    std::uint32_t workers = 0;
+    std::uint32_t target = 0;
+    std::vector<std::uint32_t> pending;
+
+    QueuedTask(SegmentBudget *b, TaskId i) : budget(b), id(i) {}
+
+    void on_target(std::uint32_t n) { pending.push_back(n); }
+
+    // Delivers the oldest queued target, applying it the way a real task's
+    // apply_slot_target()/fill_slots_locked() would: try to grow to it (confirm_slot()
+    // may deny), or shed down to it. Returns false (nothing to deliver) if the queue was
+    // empty -- the condition VT_REQUIRE checks to prove a wakeup was actually queued.
+    bool deliver_one() {
+        if (pending.empty())
+            return false;
+        target = pending.front();
+        pending.erase(pending.begin());
+        while (workers < target) {
+            if (!budget->confirm_slot(id))
+                break;
+            ++workers;
+        }
+        while (workers > target) {
+            budget->release_slot(id);
+            --workers;
+        }
+        return true;
+    }
+};
+
+}  // namespace
+
+VT_TEST(budget_release_wakes_a_denied_waiter) {
+    // Sanity baseline: max_active=1, A (higher priority) holds it, B wants one too and is
+    // fairly denied -- its target stays 0 while A outranks it and still wants its slot.
+    // Once A stops wanting one, B's target rises and B is woken via the plain
+    // target-changed path -- no wait-list needed for this simple case. The harder case
+    // below is what actually needs it.
+    SegmentBudget b({.max_active_segments = 1});
+    QueuedTask A{&b, tid(1)}, B{&b, tid(2)};
+    b.register_task(A.id, {.host = "h", .per_task_cap = 1, .resumable = true},
+                    [&](std::uint32_t n) { A.on_target(n); });
+    b.set_task_order(std::vector<TaskId>{A.id, B.id});
+    b.set_want(A.id, 1);
+    VT_REQUIRE(A.deliver_one());
+    VT_CHECK_EQ(A.workers, 1u);
+
+    // B registers and wants one too, but with A (higher priority) already holding the
+    // only slot and still wanting it, B's fairly computed target stays 0 -- unchanged
+    // from its just-registered value, so no callback is queued for it yet.
+    b.register_task(B.id, {.host = "h", .per_task_cap = 1, .resumable = true},
+                    [&](std::uint32_t n) { B.on_target(n); });
+    b.set_want(B.id, 1);
+    VT_CHECK(!B.deliver_one());
+    VT_CHECK_EQ(B.workers, 0u);
+
+    b.set_want(A.id, 0);          // A is done wanting a slot
+    VT_REQUIRE(A.deliver_one());  // A's target dropped to 0 -- sheds its held slot
+    VT_CHECK_EQ(A.workers, 0u);
+    VT_REQUIRE(B.deliver_one());  // B's target rose to 1 -- the plain target-changed path
+    VT_CHECK_EQ(B.workers, 1u);   // B took the freed slot
+}
+
+VT_TEST(budget_wait_list_wakes_a_task_whose_target_never_changed) {
+    // The real gap. max_active=2. Y alone, holds both (target=2). X arrives wanting 1:
+    // this recompute correctly drops Y's target to 1 (giving X its guaranteed slot) and
+    // raises X's target to 1 -- both real target changes, both queued. Deliver X's
+    // *first*: X's target says grow, but Y still physically holds 2 (hasn't processed
+    // its own shrink yet) -- confirm_slot() must deny X here (active_ == max_active_),
+    // which is the correctness fix (over-admission is the real RSS bug). Then Y
+    // processes its shrink and actually releases. X's target never changes again -- it
+    // was already correctly 1 -- so nothing in the plain mechanism ever revisits X.
+    SegmentBudget b({.max_active_segments = 2});
+    QueuedTask Y{&b, tid(1)}, X{&b, tid(2)};
+    b.register_task(Y.id, {.host = "h", .per_task_cap = 2, .resumable = true},
+                    [&](std::uint32_t n) { Y.on_target(n); });
+    b.set_want(Y.id, 2);
+    VT_REQUIRE(Y.deliver_one());
+    VT_CHECK_EQ(Y.workers, 2u);
+
+    b.register_task(X.id, {.host = "h", .per_task_cap = 1, .resumable = true},
+                    [&](std::uint32_t n) { X.on_target(n); });
+    b.set_task_order(std::vector<TaskId>{Y.id, X.id});
+    b.set_want(X.id, 1);
+
+    VT_REQUIRE(X.deliver_one());
+    VT_CHECK_EQ(X.workers, 0u);  // denied: active_ == max_active_, even though X's target is 1
+
+    VT_REQUIRE(Y.deliver_one());
+    VT_CHECK_EQ(Y.workers, 1u);  // Y actually releases its excess now
+
+    // The bug: without a wait-list, X.pending is empty here -- nothing was ever queued
+    // for it, because X's target never changed again. X would wait forever despite its
+    // target correctly saying it should hold a slot.
+    VT_REQUIRE(X.deliver_one());
+    VT_CHECK_EQ(X.workers, 1u);
+    VT_CHECK_EQ(b.budget().active, 2u);
+}

core/tests/task/engine_test.cpp

@@ -457,3 +457,33 @@ VT_TEST(engine_content_length_mismatch_fails_honestly) {
     VT_CHECK(rec.saw(EngineState::failed));
     VT_CHECK_EQ(::access(td.file("clm.bin").c_str(), F_OK), -1);  // never renamed into place
 }
+
+// --- DAEMON-reported bug: Progress.speed_bps reads 0 for the whole life of a live
+// download while downloaded bytes visibly advance. DAEMON reads progress by polling
+// DownloadHandle::progress() (engine_port_core.hpp), not the on_progress push callback --
+// this exercises exactly that path. ---
+
+VT_TEST(engine_polled_progress_reports_nonzero_speed) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    Recorder rec;
+    Engine eng;
+    auto h = eng.start(spec_for(srv, "/throttled/file/4M", td.file("sp.bin")), rec.cbs());
+
+    // Give it real, sustained progress: the speed estimate only updates on a >=0.5s
+    // sample window (seg_data), so a snapshot taken too early would legitimately read 0
+    // even with the bug fixed. Poll until downloaded has clearly advanced twice over.
+    std::uint64_t speed = 0;
+    for (int i = 0; i < 400 && speed == 0; ++i) {
+        std::this_thread::sleep_for(20ms);
+        auto p = h.progress();
+        if (p.downloaded >= 256u * 1024)
+            speed = p.speed_bps;
+    }
+    VT_CHECK(speed > 0);
+
+    h.cancel(/*discard_partial=*/true);
+    auto r = rec.wait();
+    VT_REQUIRE(!r.has_value());
+}

docs/adr/0017-segment-budget-admission-and-wait-list-wakeup.md

@@ -0,0 +1,102 @@
+# 17. SegmentBudget over-admission, and a wait-list to wake denied tasks
+
+Status: accepted
+
+## Context
+
+`core/docs/m7-baseline.md`'s first pass measured ~70 MB RSS for the M1/M7 DoD's "20 active
+downloads, `max_active_segments=32`" scenario, against a 60 MB target `docs/adr/0012`
+estimated would hold with ~45–50 MB of margin. `tools/bench heap-profile` (added to chase
+this down — no massif/heaptrack in this environment) found the actual cause directly:
+`Engine::segment_budget().budget().active` read 56–86 against a `total` of 32. The engine
+was not over budget by a measurement artifact or an ADR 0012 arithmetic gap; it was
+genuinely running more concurrent segments — and their 1 MiB ring buffers — than
+`max_active_segments` allows.
+
+`SegmentBudget::confirm_slot(id)` checked only `t.held < t.target` — a per-task check.
+`reallocate_locked()`'s two-pass fairness allocation does correctly bound
+`sum(target) <= max_active_` at the moment it computes a plan, but that bound says nothing
+about `sum(held)`: a task can be legitimately holding more than its own just-lowered
+target for a while, because yield is deliberately deferred to a segment boundary, never
+mid-segment (ADR 0011 A1). When that happens, another task's target can correctly rise to
+claim the capacity the first task is *about* to give back, and both `confirm_slot()` calls
+can succeed against their own, individually-correct targets while the sum of what's
+*physically held* exceeds `max_active_`. Reproduced deterministically in
+`core/tests/segment/budget_test.cpp` (`budget_active_never_exceeds_max_active_segments_
+under_concurrent_load`, `budget_wait_list_wakes_a_task_whose_target_never_changed`)
+independent of any real network I/O.
+
+## Decision
+
+**`confirm_slot()` also checks `active_ < max_active_`, unconditionally**, as a backstop
+that doesn't depend on any task's target bookkeeping being perfectly in sync with what
+every other task currently holds. This is the fix for the invariant itself.
+
+That backstop creates a liveness question the original design never had to answer: a task
+denied *only* by this new check has a target that's already correct — it doesn't change
+again on its own, so `reallocate_locked()`'s plain "fire a callback when a task's target
+changes" mechanism never revisits it. Nothing was watching for "this specific task is owed
+a slot"; **`Task` gained a `waiting_for_slot` flag**, set by `confirm_slot()` on exactly
+this denial and cleared on the task's next successful confirm however that retry was
+triggered. `release_slot()` and `deregister_task()` — the only two places that can free
+real capacity — now call `wake_one_waiter_locked()`, which hands the newly-freed slot to
+the highest-priority `waiting_for_slot` task (other than the one that just released) via a
+direct retry hint, if `reallocate_locked()`'s own plan didn't already produce one.
+
+On the `download_task.cpp` side, a woken task's `apply_slot_target()` → `fill_slots_locked()`
+also needed a fix: a segment that had backed off mid-retry (`SegState::stalled`, its
+`release_slot()` already called, a `retry_worker()` timer already scheduled and possibly
+already denied once) has no live worker and never surfaces through
+`Segmenter::assign_slot()` — it stays *assigned*, just not running, and assign_slot() only
+ever hands out unassigned or fresh ranges. `fill_slots_locked()` now restarts any stalled
+segment with no live worker directly (bounded by `slot_target`, same as its `assign_slot()`
+loop) before looking for new work; a segment it doesn't get to keeps its own scheduled
+`retry_worker()` timer as a second chance, so this is additive, not a replacement for that
+path.
+
+### What this isn't
+
+Production's real `on_target` callback (`register_task()`'s lambda, `download_task.cpp`)
+never runs synchronously with whatever budget call triggered it — it posts through
+`host.schedule()` (`engine.cpp`'s single timer thread), so a task's own `mu` can never be
+re-entered on the same call stack, and two tasks' callbacks can never race each other into
+an AB-BA lock order either (only one ever runs at a time, on one thread). An earlier
+version of this fix tried to defend against a same-thread reentrancy hazard that,
+diagnosed correctly, doesn't exist in production at all — it exists only if a *test double*
+calls back into the budget synchronously from inside `on_target`, which none of production
+does. `core/tests/segment/budget_test.cpp`'s `FakeTask` does exactly that (by design — it's
+simple and every other test in the file drives the budget from one thread at a time, where
+that's harmless); the one test that drives it from *multiple* concurrent threads
+(`budget_concurrent_confirm_release_stays_consistent`) uses a separate `AsyncFakeTask` that
+posts through a small `TestTimer`, mirroring `host.schedule()` / the engine's timer thread
+for real, rather than adding synchronization machinery to `SegmentBudget` itself to paper
+over a test double being unlike production. `wake_one_waiter_locked()`'s `exclude`
+parameter is the one piece of that defense that's independently justified either way — a
+task doesn't need to be told to retry a slot it just gave back itself — and is the only
+piece that stayed.
+
+## Consequences
+
+- The M7 RSS number now clears the DoD line: `heap-profile` reports 45.41 MiB for the
+  20-task/8-segment/32-cap scenario (`core/docs/m7-baseline.md`), and
+  `budget.active` never exceeds `budget.total` regardless of contention.
+- `docs/adr/0016`'s TSan-only load-test straggler (a *different* subsystem —
+  `rate::RateLimiter`'s byte-pacing, not `SegmentBudget`'s segment-admission) is now
+  suspected to have been this same root cause rather than the `CURLOPT_LOW_SPEED_TIME`
+  guess offered there, since the symptom (a task that simply never resumes) matches
+  exactly. Not reverified at the DoD's full shape under `--preset tsan` in this change —
+  `tools/bench`'s sanitizer-preset `load` registration still runs at reduced concurrency.
+  Worth a follow-up run before closing that ADR's postscript.
+- `docs/adr/0016`'s actual subject — `rate::RateLimiter::TokenBucket`'s own peek/commit
+  race, unrelated to `SegmentBudget` — is untouched by this change and remains open.
+- `wake_one_waiter_locked()` wakes exactly one task per actual release, by priority order.
+  Under sustained heavy oversubscription (this bench's 20 tasks × 8 segments = 160 wanted
+  against 32 available) a low-priority task can still wait a long time for its fair share
+  — that's the two-pass allocator's own fairness policy working as designed, not a
+  liveness bug: it will get there, just not fast, and every fresh `reallocate_locked()`
+  call (any task's `set_want`, registration, or departure) reconsiders everyone from
+  scratch. No test in this change measures *how* long; if that ever needs a stronger
+  guarantee (bounded wait time, not just eventual service), it's a fairness-policy change,
+  not a wiring fix.
+
+Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

tools/bench/CMakeLists.txt

@@ -37,17 +37,18 @@ if(VELOX_BUILD_TESTS)
     set_tests_properties(vdm_bench_throughput_smoke PROPERTIES LABELS "bench" TIMEOUT 120)
 
     # --tasks 8 --segments 2 (not the DoD's 20 tasks * default_segments=8 = 160 concurrent
-    # segments): at the full shape, TSan's per-access instrumentation overhead was observed
+    # segments): at the full shape under --preset tsan, a straggler task was observed not
-    # to leave a straggler task not just slow but still incomplete past a 300s-per-task
+    # just slow but still incomplete past a 300s-per-task budget (docs/adr/0016's
-    # budget -- reproduced at tasks=20/segments=8 (2 stragglers) and, smaller but still
+    # postscript, written before docs/adr/0017's SegmentBudget fix landed) -- now suspected
-    # present, at tasks=20/segments=2 (1 straggler); tasks=8/segments=2 (16 concurrent
+    # to have been that same over-admission bug (a segment denied a slot with nothing to
-    # connections) was reliable across repeated runs. No TSan report ever accompanied a
+    # wake it), not a TSan-timing artifact as first guessed, since the symptom -- a task
-    # straggler (this isn't a race -- see docs/adr/0016's postscript), so it reads as some
+    # that simply never resumes -- matches exactly. Left at this reduced concurrency rather
-    # combination of TSan's overhead and this environment's scheduling, not an engine bug;
+    # than reverting: re-verifying the full shape is clean under TSan wasn't done as part
-    # still, "every task completes" is exactly what this smoke test is supposed to check
+    # of that fix (see docs/adr/0017's "Consequences"), so this is still the bar that's
-    # (see the split above), so the bar it runs at has to be one that actually holds. The
+    # actually known to hold. "Every task completes" is what this smoke test exists to
-    # DoD's real 20-task/default-segments/60MB-RSS shape is exercised by the manual/CI M7
+    # check (see the split above); the DoD's real 20-task/default-segments/60MB-RSS shape
-    # sign-off run in this file's header comment, at --preset release, where it passes.
+    # is exercised by the manual/CI M7 sign-off run in this file's header comment, at
+    # --preset release, where it passes (core/docs/m7-baseline.md).
     add_test(NAME vdm_bench_load20 COMMAND vdm_bench load --tasks 8 --task-size 2M --segments 2)
     set_tests_properties(vdm_bench_load20 PROPERTIES LABELS "bench" TIMEOUT 300)
 

tools/bench/vdm_bench.cpp

@@ -22,6 +22,18 @@
 //       for a --preset release run, where the number means something (a sanitizer roughly
 //       doubles-to-quadruples RSS via redzones/shadow memory).
 //
+//   vdm_bench heap-profile [--tasks 20] [--task-size 4M] [--segments N] [--top N]
+//       Answers "where does the RSS actually go", for the M7 RSS target (docs/04 §8) --
+//       no massif/heaptrack in this environment (no root to install them), so this is a
+//       small in-process equivalent: the same operator new/delete override tracks, per
+//       allocation, which call site made it (one return address, via
+//       __builtin_return_address(0) right in operator new -- cheap enough to run for the
+//       whole scenario, unlike a full backtrace) and keeps a live (not-yet-freed) byte
+//       count per site. Runs the same 20-task concurrent scenario as `load`, waits for
+//       peak RSS to stop growing, then prints the top sites by live bytes -- the
+//       breakdown ADR 0012's "45-50 MB" estimate needs to be checked against, and
+//       core/docs/m7-baseline.md's ~70 MB measurement needs explained.
+//
 //   vdm_bench alloc-check [--size 128M] [--window-s 2]
 //       docs/agents/AGENT-CORE.md: "no allocation in the curl write callback... checked in
 //       review and by a bench assertion." operator new/delete are overridden process-wide
@@ -42,17 +54,22 @@
 
 #include "vdm/engine.hpp"
 
+#include <cxxabi.h>
+#include <dlfcn.h>
 #include <sys/resource.h>
 #include <unistd.h>
 
+#include <algorithm>
 #include <atomic>
 #include <chrono>
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
 #include <future>
+#include <mutex>
 #include <new>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #include "support/local_server.hpp"
@@ -62,20 +79,194 @@ using namespace vdm;
 using namespace vdm::task;
 using namespace std::chrono_literals;
 
-// --- allocation counting (alloc-check only; harmless overhead otherwise) --------------
+// --- allocation tracking: a plain counter for alloc-check, and (only when heap-profile
+// turns it on) a per-call-site live-byte tracker standing in for massif/heaptrack, neither
+// of which is installable here (no root). Both share one operator new/delete override
+// since a process gets exactly one. ---------------------------------------------------
 
 namespace {
 std::atomic<std::uint64_t> g_alloc_count{0};
+
+std::atomic<bool> g_site_tracking{false};
+// Re-entrancy guard: our own bookkeeping below calls into std::unordered_map, which
+// allocates. Without this, that nested operator new call would recurse into the same
+// tracking logic -- not infinitely (the nested call's own bookkeeping call would itself be
+// guarded the same way, so it terminates), but it would count the tracker's own node
+// allocations as if they were the program's, polluting the profile. When set, an
+// allocation is just satisfied via malloc/free with no bookkeeping.
+thread_local bool g_in_tracker = false;
+
+struct LiveAlloc {
+    std::size_t size;
+    void *site;  // __builtin_return_address(0) at the call to operator new
+};
+struct SiteStats {
+    std::uint64_t live_bytes = 0;
+    std::uint64_t alloc_count = 0;  // lifetime, not just live -- how *often* a site fires
+};
+
+std::mutex g_track_mu;
+std::unordered_map<void *, LiveAlloc> g_live;         // pointer -> what/where
+std::unordered_map<void *, SiteStats> g_site_stats;    // call site -> aggregate
+
+void track_alloc(void *p, std::size_t n) {
+    if (!g_site_tracking.load(std::memory_order_relaxed) || g_in_tracker)
+        return;
+    g_in_tracker = true;
+    void *site = __builtin_return_address(0);
+    {
+        std::lock_guard lk(g_track_mu);
+        g_live.emplace(p, LiveAlloc{n, site});
+        auto &st = g_site_stats[site];
+        st.live_bytes += n;
+        st.alloc_count += 1;
+    }
+    g_in_tracker = false;
 }
 
+void track_free(void *p) {
+    if (!p || !g_site_tracking.load(std::memory_order_relaxed) || g_in_tracker)
+        return;
+    g_in_tracker = true;
+    {
+        std::lock_guard lk(g_track_mu);
+        if (auto it = g_live.find(p); it != g_live.end()) {
+            g_site_stats[it->second.site].live_bytes -= it->second.size;
+            g_live.erase(it);
+        }
+    }
+    g_in_tracker = false;
+}
+
+std::string describe_site(void *site);  // defined below; used as a per-site fallback here
+
+// Resolves a batch of call-site addresses to "function (file:line)" via one `addr2line`
+// invocation (reads .symtab + DWARF, unlike dladdr below -- most of this codebase's
+// call sites are internal-linkage or file-static and never make it into the *dynamic*
+// symbol table dladdr is limited to). `sites` and the returned vector are in the same
+// order. dladdr's `dli_fbase` (the object's runtime load base -- works even when it can't
+// name the specific symbol, which is exactly the "site not in .dynsym" case this exists
+// for) converts each ASLR'd runtime address into the file-relative offset addr2line wants.
+// Falls back to a raw "0x..." string per site if addr2line isn't on PATH or the batch call
+// otherwise fails -- describe_site() below still has dladdr as a second-line fallback for
+// an individual site in that case.
+std::vector<std::string> resolve_sites_via_addr2line(const std::vector<void *> &sites) {
+    std::vector<std::string> out(sites.size());
+    for (std::size_t i = 0; i < sites.size(); ++i) {
+        char buf[32];
+        std::snprintf(buf, sizeof buf, "%p", sites[i]);
+        out[i] = buf;
+    }
+    if (sites.empty())
+        return out;
+
+    // `-e /proc/self/exe` would resolve inside the *addr2line* process this spawns, i.e.
+    // to addr2line's own binary, not vdm_bench's -- readlink it here, in-process, first.
+    char exe_path[4096];
+    ssize_t exe_len = ::readlink("/proc/self/exe", exe_path, sizeof(exe_path) - 1);
+    if (exe_len <= 0)
+        return out;
+    exe_path[exe_len] = '\0';
+
+    std::string cmd = std::string("addr2line -f -C -e '") + exe_path + "'";
+    for (void *site : sites) {
+        Dl_info info{};
+        std::uintptr_t off = reinterpret_cast<std::uintptr_t>(site);
+        if (::dladdr(site, &info) && info.dli_fbase)
+            off -= reinterpret_cast<std::uintptr_t>(info.dli_fbase);
+        char buf[24];
+        std::snprintf(buf, sizeof buf, " %zx", off);
+        cmd += buf;
+    }
+    cmd += " 2>/dev/null";
+
+    FILE *pipe = ::popen(cmd.c_str(), "r");
+    if (!pipe)
+        return out;
+    std::vector<std::string> lines;
+    char line[1024];
+    while (std::fgets(line, sizeof line, pipe)) {
+        std::string s = line;
+        while (!s.empty() && (s.back() == '\n' || s.back() == '\r')) s.pop_back();
+        lines.push_back(std::move(s));
+    }
+    ::pclose(pipe);
+
+    // addr2line -f prints two lines per address: function name, then file:line.
+    for (std::size_t i = 0; i * 2 + 1 < lines.size() && i < sites.size(); ++i) {
+        const std::string &fn = lines[i * 2];
+        const std::string &loc = lines[i * 2 + 1];
+        if (fn == "??" && loc == "??:0")
+            out[i] = describe_site(sites[i]);  // addr2line drew a blank -- try dladdr
+        else
+            out[i] = fn + " (" + loc + ")";
+    }
+    return out;
+}
+
+// Single-site fallback (dladdr only) for when addr2line can't place an address (it read
+// only .symtab -- e.g. no debug/symbol data at all for that address) or isn't available.
+std::string describe_site(void *site) {
+    Dl_info info{};
+    if (!::dladdr(site, &info) || !info.dli_sname) {
+        char buf[32];
+        std::snprintf(buf, sizeof buf, "%p", site);
+        return buf;
+    }
+    std::string name = info.dli_sname;
+    int status = 0;
+    if (char *dem = abi::__cxa_demangle(name.c_str(), nullptr, nullptr, &status);
+        dem && status == 0) {
+        name = dem;
+        std::free(dem);
+    }
+    auto off = static_cast<std::uintptr_t>(reinterpret_cast<char *>(site) -
+                                           reinterpret_cast<char *>(info.dli_saddr));
+    return name + " +0x" + [&] {
+        char buf[24];
+        std::snprintf(buf, sizeof buf, "%zx", off);
+        return std::string(buf);
+    }();
+}
+
+}  // namespace
+
 void *operator new(std::size_t n) {
     g_alloc_count.fetch_add(1, std::memory_order_relaxed);
-    if (void *p = std::malloc(n ? n : 1))
+    void *p = std::malloc(n ? n : 1);
-        return p;
+    if (!p)
-    throw std::bad_alloc();
+        throw std::bad_alloc();
+    track_alloc(p, n);
+    return p;
+}
+// The nothrow overload is a distinct allocation function from the plain one above, not an
+// alternate spelling of it: leaving it un-overridden means it falls through to ASan's own
+// default (nothrow) operator new, while a deallocation of that same pointer still reaches
+// the plain `operator delete` override below -- a real alloc-dealloc-mismatch ASan catches
+// (hit via std::stable_sort's std::get_temporary_buffer, which allocates this way; see
+// core/src/segment/budget.cpp's SegmentBudget::run() for why that path no longer uses
+// stable_sort at all, but this override is fixed too since anything else pulling in a
+// nothrow allocation -- directly or via another std:: temp-buffer user -- would hit it
+// again otherwise).
+void *operator new(std::size_t n, const std::nothrow_t &) noexcept {
+    g_alloc_count.fetch_add(1, std::memory_order_relaxed);
+    void *p = std::malloc(n ? n : 1);
+    if (p)
+        track_alloc(p, n);
+    return p;
+}
+void operator delete(void *p) noexcept {
+    track_free(p);
+    std::free(p);
+}
+void operator delete(void *p, std::size_t) noexcept {
+    track_free(p);
+    std::free(p);
+}
+void operator delete(void *p, const std::nothrow_t &) noexcept {
+    track_free(p);
+    std::free(p);
 }
-void operator delete(void *p) noexcept { std::free(p); }
-void operator delete(void *p, std::size_t) noexcept { std::free(p); }
 
 namespace {
 
@@ -233,6 +424,42 @@ int cmd_throughput(const Args &a) {
     return rc;
 }
 
+const char *seg_state_name(segment::SegState s) {
+    switch (s) {
+        case segment::SegState::idle: return "idle";
+        case segment::SegState::connecting: return "connecting";
+        case segment::SegState::downloading: return "downloading";
+        case segment::SegState::stalled: return "stalled";
+        case segment::SegState::complete: return "complete";
+        case segment::SegState::failed: return "failed";
+    }
+    return "?";
+}
+
+// Printed once, on a task timeout, for whatever hang it wasn't designed to reproduce --
+// per task: engine state, worker/segment count, and every segment's own state; plus the
+// budget's own view, engine-wide. Enough to tell "stuck starved" (budget.active at cap,
+// nobody holding this task's segments) from "stuck stalled" (a segment sitting in
+// `stalled` with no worker, budget has room) from anything else, without re-deriving it by
+// re-running the bench.
+void dump_state(Engine &eng, const std::vector<DownloadHandle> &handles, int timed_out_task) {
+    std::fprintf(stderr, "---- state dump (task %d timed out) ----\n", timed_out_task);
+    auto eb = eng.segment_budget().budget();
+    std::fprintf(stderr, "budget: total=%u active=%u starved=%u\n", eb.total, eb.active,
+                eb.tasks_starved);
+    for (std::size_t i = 0; i < handles.size(); ++i) {
+        auto p = handles[i].progress();
+        std::fprintf(stderr, "task %zu: state=%d downloaded=%llu/%llu effective_segments=%u\n",
+                    i, static_cast<int>(handles[i].state()), (unsigned long long)p.downloaded,
+                    (unsigned long long)p.total.value_or(0), p.effective_segments);
+        for (auto &sp : p.segments)
+            std::fprintf(stderr, "  seg %u: state=%s completed=%llu speed_bps=%llu\n", sp.index,
+                        seg_state_name(sp.state), (unsigned long long)sp.completed,
+                        (unsigned long long)sp.speed_bps);
+    }
+    std::fprintf(stderr, "----------------------------------------\n");
+}
+
 int cmd_load(const Args &a) {
     const int tasks = static_cast<int>(a.get_long("--tasks", 20));
     const std::uint64_t task_size = a.get_size("--task-size", "4M");
@@ -310,6 +537,7 @@ int cmd_load(const Args &a) {
     for (int i = 0; i < tasks; ++i) {
         if (futs[i].wait_for(task_timeout) != std::future_status::ready) {
             std::fprintf(stderr, "task %d: timed out\n", i);
+            dump_state(eng, handles, i);
             ++failures;
             continue;
         }
@@ -336,6 +564,102 @@ int cmd_load(const Args &a) {
     return 0;
 }
 
+int cmd_heap_profile(const Args &a) {
+    const int tasks = static_cast<int>(a.get_long("--tasks", 20));
+    const std::uint64_t task_size = a.get_size("--task-size", "4M");
+    const auto segments_override = static_cast<std::uint32_t>(a.get_long("--segments", 0));
+    const int top_n = static_cast<int>(a.get_long("--top", 20));
+
+    std::string dir = tmp_workdir();
+    const std::uint64_t assumed_segments = segments_override ? segments_override : 8;
+    const std::uint64_t per_conn_bps =
+        std::max<std::uint64_t>(1, task_size / (5 * assumed_segments));
+    vdm::bench::TestServerProc srv(per_conn_bps);
+    if (!srv.available()) {
+        std::fprintf(stderr,
+                     "vdm_bench: tools/testserver unavailable -- skipping heap-profile.\n");
+        return 0;
+    }
+
+    // On from before Engine's own construction, so its fixed cost (http worker threads,
+    // curl_multi handles, ...) shows up in the profile too, not just what the tasks add.
+    g_site_tracking.store(true, std::memory_order_relaxed);
+    Engine eng;
+
+    std::vector<DownloadHandle> handles;
+    handles.reserve(tasks);
+    for (int i = 0; i < tasks; ++i) {
+        DownloadSpec spec;
+        spec.url = srv.url("/throttled/file/" + std::to_string(task_size));
+        spec.save_path = dir + "/out" + std::to_string(i) + ".bin";
+        if (segments_override)
+            spec.segments = segments_override;
+        handles.push_back(eng.start(std::move(spec), DownloadCallbacks{}));
+    }
+
+    // Wait for peak RSS to stop growing (5 consecutive 200ms samples with no increase, or
+    // a generous cap) -- that's the moment whose live-allocation breakdown answers "where
+    // does the RSS actually go", the same question core/docs/m7-baseline.md's ~70 MB
+    // number raises against ADR 0012's ~45-50 MB estimate.
+    long last_rss = 0, stable_for = 0;
+    for (int i = 0; i < 300 && stable_for < 5; ++i) {
+        std::this_thread::sleep_for(200ms);
+        long rss = sample_rusage().peak_rss_kb;
+        stable_for = (rss <= last_rss) ? stable_for + 1 : 0;
+        last_rss = rss;
+    }
+
+    // Cross-check against the budget's own accounting, independent of the allocator-site
+    // snapshot below: this is what caught the real find here (core/docs/m7-baseline.md,
+    // docs/adr/0012) -- sum(effective_segments) tracked budget.active well past
+    // budget.total before the SegmentBudget fix these numbers are now reported alongside.
+    std::uint32_t total_effective_segments = 0;
+    for (auto &h : handles) total_effective_segments += h.progress().effective_segments;
+    auto eb = eng.segment_budget().budget();
+    std::fprintf(stderr,
+                "segment budget: %u live across tasks, engine reports total=%u active=%u "
+                "starved=%u\n",
+                total_effective_segments, eb.total, eb.active, eb.tasks_starved);
+
+    // Turn tracking off *before* touching g_site_stats: the snapshot below allocates its
+    // own vector storage through the very same overridden operator new. With tracking
+    // still on and g_track_mu already held for the snapshot, that allocation would recurse
+    // into track_alloc() and try to lock g_track_mu again -- self-deadlock on a
+    // non-recursive mutex (hit this exact hang while first wiring this subcommand up:
+    // g_in_tracker only guards the tracker's own bookkeeping containers, not other code
+    // that allocates while holding g_track_mu directly).
+    g_site_tracking.store(false, std::memory_order_relaxed);
+    std::vector<std::pair<void *, SiteStats>> sites;
+    {
+        std::lock_guard lk(g_track_mu);
+        sites.assign(g_site_stats.begin(), g_site_stats.end());
+    }
+    std::sort(sites.begin(), sites.end(), [](const auto &x, const auto &y) {
+        return x.second.live_bytes > y.second.live_bytes;
+    });
+
+    std::fprintf(stderr, "heap-profile: peak RSS %.2f MiB, %d tasks, %llu segments assumed\n",
+                last_rss / 1024.0, tasks, (unsigned long long)assumed_segments);
+    std::fprintf(stderr, "%-10s %-10s  %s\n", "live KiB", "#allocs", "call site");
+    std::uint64_t total_live = 0;
+    for (auto &[site, st] : sites) total_live += st.live_bytes;
+    std::fprintf(stderr, "total tracked live: %.2f MiB across %zu sites\n\n",
+                total_live / 1024.0 / 1024.0, sites.size());
+    std::vector<void *> top_addrs;
+    for (int i = 0; i < top_n && i < static_cast<int>(sites.size()); ++i) top_addrs.push_back(sites[i].first);
+    auto labels = resolve_sites_via_addr2line(top_addrs);
+    for (std::size_t i = 0; i < top_addrs.size(); ++i) {
+        auto &[site, st] = sites[i];
+        std::fprintf(stderr, "%-10.1f %-10llu  %s\n", st.live_bytes / 1024.0,
+                    (unsigned long long)st.alloc_count, labels[i].c_str());
+    }
+
+    // Handles/Engine go out of scope here: ~Engine quiesces every task (waits for its
+    // workers to actually drain -- see the quiesce() fix) rather than waiting for the
+    // downloads to run to completion, which isn't the point of this subcommand.
+    return 0;
+}
+
 int cmd_alloc_check(const Args &a) {
     const std::uint64_t size = a.get_size("--size", "128M");
     const double window_s = a.get_double("--window-s", 2.0);
@@ -416,12 +740,13 @@ int cmd_alloc_check(const Args &a) {
 
 void usage() {
     std::fprintf(stderr,
-                 "usage: vdm_bench <throughput|load|alloc-check> [options]\n"
+                 "usage: vdm_bench <throughput|load|heap-profile|alloc-check> [options]\n"
-                 "  throughput [--size 5G] [--segments N] [--require-mbps N] "
+                 "  throughput   [--size 5G] [--segments N] [--require-mbps N] "
                  "[--max-cpu-pct N]\n"
-                 "  load       [--tasks 20] [--task-size 4M] [--segments N] "
+                 "  load         [--tasks 20] [--task-size 4M] [--segments N] "
                  "[--require-rss-kb N] [--task-timeout-s 300]\n"
-                 "  alloc-check [--size 128M] [--window-s 2] [--budget-per-s 5]\n");
+                 "  heap-profile [--tasks 20] [--task-size 4M] [--segments N] [--top 20]\n"
+                 "  alloc-check  [--size 128M] [--window-s 2] [--budget-per-s 5]\n");
 }
 
 }  // namespace
@@ -437,6 +762,8 @@ int main(int argc, char **argv) {
         return cmd_throughput(args);
     if (cmd == "load")
         return cmd_load(args);
+    if (cmd == "heap-profile")
+        return cmd_heap_profile(args);
     if (cmd == "alloc-check")
         return cmd_alloc_check(args);
     usage();