core: rate/token_bucket + fold in DAEMON's engine-API review (stage 7)

rate/token_bucket.hpp — a lazily-refilled TokenBucket (starts full: burst
then throttle, IDM behaviour; rate 0 == unlimited; burst caps idle
accumulation) and RateLimiter, the global -> per-queue -> per-task
hierarchy (docs/04 §6). acquire(task, n) peeks every applicable level and
commits on all-or-none so a blocked attempt never leaks tokens at a level
that had them; held under one mutex so a concurrent detach can't dangle
the bucket it's using. vdm/ids.hpp gains QueueId.

Tests: burst/refill/cap/unlimited for the bucket; tightest-level-binds,
no-partial-consumption, detach-safety, and an 8-thread aggregate-rate
check for the hierarchy. Green under ASan/UBSan and TSan.

Engine-API review (DAEMON signed off, no sched/ or dispatch rewrite):
 - Engine::rate_limiter() accessor added (limiter.set -> set_global_limit).
 - Checksum::Algo gains sha512 to match the wire Checksum set.
 - DownloadSpec: DAEMON creates save_path's parent dir before start();
   missing dir -> Error::path_rejected (made explicit).
 - cancel(): documented to fire on_state(_, cancelled, nullopt) then
   on_finished(Err{canceled}), in that order; download.cancel ==
   cancel(false), download.remove == cancel(true).
 - engine-api-m1.md: the five open questions resolved with DAEMON's
   answers (probe_hint optional, single cancel flag, {restart,
   keep_partial, abort} is the whole set, per-task 4 Hz is fine,
   refresh_url restarts all segments after a validating re-probe).

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

core/docs/engine-api-m1.md

@@ -1,6 +1,18 @@
-# CORE → DAEMON — the engine API for review (M1)
+# CORE → DAEMON — the engine API (M1)
 
-**Status: for review.** This is the `libveloxcore` public surface `veloxd` links and calls
+**Status: reviewed and signed off** by DAEMON on 2026-09-10
+(`daemon/docs/engine-api-review.md`). Nothing forced a `sched/` or dispatch rewrite. The
+five open questions are resolved at the bottom; the review's four confirms are folded into
+the headers (`sha512` added to `Checksum::Algo`; the parent-directory contract made
+explicit; `cancel()` ordering documented; `rate_limiter()` accessor added).
+
+Integration order: DAEMON wires this in after `daemon/src/sched/` lands (the scheduler is
+what calls `start`/`pause`/`resume`/`cancel` and drives `set_task_order`); `sched/` builds
+against these headers in parallel with CORE stage 8 and does not need the bodies.
+
+---
+
+This is the `libveloxcore` public surface `veloxd` links and calls
 to actually run a download. It is the thing the AGENT-CORE brief asked for on day one and
 that slipped: DAEMON has an RPC surface and a store and, until this is agreed, nothing to
 call. Sketch headers: `core/include/vdm/engine.hpp`, `core/include/vdm/task/download.hpp`
@@ -109,9 +121,15 @@ task waiting on credentials.
 - **`decide(Decision)`** — acts only while auto-paused for a decision. `restart` discards
   the partial and re-downloads; `keep_partial` continues against what is on disk (the
   user's stated risk); `abort` → `failed`. No-op otherwise.
-- **`refresh_url(url, headers)`** — IDM's "Refresh Download Address": swap the URL on a
-  live or paused task without losing progress (a fresh signed URL). Maps to
-  `download.refreshUrl`.
+- **`refresh_url(url, headers)`** — IDM's "Refresh Download Address": the engine re-probes
+  the new URL to validate it, then **restarts every segment on it** (the signed-URL-expiry
+  case the wire method exists for), keeping the bytes already on disk. Mirror rotation is
+  *not* this — that is `spec.mirrors` + the segmenter's requeue-to-a-different-host.
+- **`on_decision_needed`** is only for the cases the engine cannot resolve itself. A
+  routine 416 / stale range is the engine's own re-probe + re-split loop; it escalates
+  `DecisionRequest{range_metadata_stale}` **only when that loop fails**, at which point
+  "retry the same range" is already exhausted — so `{restart, keep_partial, abort}` is the
+  whole choice set.
 
 ## 5. Progress
 
@@ -138,21 +156,31 @@ source until the stream ends.
 
 ---
 
-## Open questions for DAEMON
+## Resolved (DAEMON review, 2026-09-10)
 
-1. **`DownloadSpec.probe_hint`** — do you want to pass the File-Info `ProbeResult` in, or
-   would you rather the engine always probe (one code path, ~1 extra round trip)? The
-   sketch supports both; picking one simplifies stage 8.
-2. **`cancel(discard_partial)`** vs a separate `handle.remove()` — the wire has
-   `download.cancel` and `download.remove {deleteFile}` as two methods. One call with a
-   flag, or two?
-3. **`on_decision_needed` granularity** — is `{restart, keep_partial, abort}` the right
-   choice set for `server_file_changed`, or do you also need "retry the same range once
-   more" as a distinct option for the stale-416 case?
-4. **Progress cadence** — 4 Hz per task matches the wire. With 20 active tasks that is 80
-   callbacks/s on the engine's timer thread. Acceptable, or do you want a single
-   `on_progress_batch(span<Progress>)` so the engine coalesces across tasks too?
-5. **`refresh_url` while `downloading`** — should an in-flight segment finish on the old
-   URL and only new/retried segments use the new one (less disruption), or should the
-   engine restart all segments on the new URL immediately (guaranteed consistency)? The
-   signed-URL-expiry case wants the latter; a mirror swap wants the former.
+1. **`probe_hint` stays optional.** DAEMON has a `ProbeResult` only on the File-Info path;
+   capture-take, `velox add`, `addBatch` and restart have none. The engine probes when
+   it's absent.
+2. **One `cancel(discard_partial)`.** `download.cancel` = `cancel(false)`;
+   `download.remove` = `cancel(true)` for a live task (plus DAEMON's row/file cleanup), or
+   pure DAEMON-side for an already-terminal one. No `handle.remove()`.
+3. **`{restart, keep_partial, abort}` is the whole set.** The engine owns the routine
+   416 re-probe/re-split and only escalates `on_decision_needed` when that loop fails —
+   "retry the same range" is exhausted by then.
+4. **Per-task 4 Hz is fine.** DAEMON coalesces across tasks for `event.task.progress`
+   regardless; `on_progress_batch` is a nice-to-have and must not block stage 8.
+5. **`refresh_url` restarts all segments on the new URL** after a validating re-probe (the
+   signed-URL case). Mirror rotation is `spec.mirrors` + the segmenter, not this.
+
+## Review confirms, folded in
+
+- **(a)** `vdm::TaskId` is a cheap-copy hashable value; DAEMON never constructs one — it
+  only receives it from `start()` / callbacks and passes it back to `set_task_order()`
+  etc. ✔ (`vdm/ids.hpp`)
+- **(b)** DAEMON `mkdir -p`s `save_path`'s parent before `start()`. The engine opens the
+  file and fails with `Error::path_rejected` if the directory is missing. ✔ (documented on
+  `DownloadSpec`)
+- **(c)** `Checksum::Algo` now has `sha512`, matching the wire `Checksum` set. ✔
+- **(d)** `cancel()` always fires `on_state(_, cancelled, nullopt)` then
+  `on_finished(Err{Error::canceled})` (note: the taxonomy value is `canceled`), in that
+  order. ✔ (documented on `DownloadHandle::cancel`)

core/include/vdm/engine.hpp

@@ -13,6 +13,7 @@
 #include <cstdint>
 #include <memory>
 
+#include "vdm/rate/token_bucket.hpp"
 #include "vdm/segment/budget.hpp"
 #include "vdm/task/download.hpp"
 
@@ -53,6 +54,12 @@ class Engine {
     // on_budget_changed). See ADR 0011.
     [[nodiscard]] segment::SegmentBudget &segment_budget() noexcept;
 
+    // The hierarchical speed limiter (docs/04 §6): global -> per-queue -> per-task token
+    // buckets. `limiter.set {globalBps, enabled}` -> rate_limiter().set_global_limit();
+    // per-queue / per-task limits and the task<->queue attachment come from DAEMON too.
+    // The engine paces every segment read through it.
+    [[nodiscard]] rate::RateLimiter &rate_limiter() noexcept;
+
     // Live settings (connection.* changes from settings.set). Each affects future work.
     void set_default_segments(std::uint32_t n);
     void set_default_buffer_bytes(std::uint64_t bytes);

core/include/vdm/ids.hpp

@@ -24,6 +24,15 @@ struct TaskId {
     friend constexpr auto operator<=>(const TaskId &, const TaskId &) = default;
 };
 
+// A scheduler queue. A task may belong to one — for the per-queue rate limit and
+// per-queue concurrency. A task with no queue rate-limits against the global bucket only.
+struct QueueId {
+    std::uint64_t value = 0;
+
+    [[nodiscard]] constexpr bool valid() const noexcept { return value != 0; }
+    friend constexpr auto operator<=>(const QueueId &, const QueueId &) = default;
+};
+
 using SteadyTime = std::chrono::steady_clock::time_point;
 
 }  // namespace vdm
@@ -35,4 +44,11 @@ struct std::hash<vdm::TaskId> {
     }
 };
 
+template <>
+struct std::hash<vdm::QueueId> {
+    std::size_t operator()(vdm::QueueId id) const noexcept {
+        return std::hash<std::uint64_t>{}(id.value);
+    }
+};
+
 #endif  // VDM_IDS_HPP

core/include/vdm/rate/token_bucket.hpp

@@ -0,0 +1,186 @@
+// vdm/rate/token_bucket.hpp — a lazily-refilled token bucket, and the global -> queue ->
+// task limiter hierarchy built on it (docs/04 §6).
+//
+// A segment worker calls RateLimiter::acquire(task, n) after receiving n body bytes. If
+// every applicable level (task, its queue, global) has n tokens, it consumes n from each
+// and returns 0. Otherwise it consumes nothing and returns how long to wait before
+// retrying — the worker returns CURL_WRITEFUNC_PAUSE and schedules a curl_easy_pause
+// resume after that delay (the "precision" layer on top of CURLOPT_MAX_RECV_SPEED_LARGE).
+//
+// This header compiles standalone.
+
+#ifndef VDM_RATE_TOKEN_BUCKET_HPP
+#define VDM_RATE_TOKEN_BUCKET_HPP
+
+#include <algorithm>
+#include <chrono>
+#include <cstdint>
+#include <mutex>
+#include <optional>
+#include <unordered_map>
+
+#include "vdm/ids.hpp"
+
+namespace vdm::rate {
+
+// rate_bps == 0 means unlimited: consume() always succeeds and never waits.
+class TokenBucket {
+  public:
+    TokenBucket() = default;
+    // `burst` caps how many tokens accumulate while idle; 0 => 1 second's worth.
+    explicit TokenBucket(std::uint64_t rate_bps, std::uint64_t burst = 0) {
+        set_rate(rate_bps, burst);
+    }
+
+    void set_rate(std::uint64_t rate_bps, std::uint64_t burst = 0) {
+        std::lock_guard lk(mu_);
+        const bool was_unlimited = rate_ == 0;
+        rate_ = rate_bps;
+        cap_ = burst ? burst : rate_bps;  // 1 s of burst by default
+        // A freshly-limited bucket starts full: you may transfer at burst speed
+        // immediately, then it throttles (classic token bucket / IDM behaviour). Lowering
+        // an existing limit only clamps down — it never hands out a fresh burst.
+        if (rate_bps > 0 && was_unlimited)
+            tokens_ = cap_;
+        else if (tokens_ > cap_)
+            tokens_ = cap_;
+        last_ = clock::now();
+    }
+    [[nodiscard]] std::uint64_t rate() const {
+        std::lock_guard lk(mu_);
+        return rate_;
+    }
+
+    // Consume `n` if available; otherwise consume nothing. Returns the wait until `n`
+    // tokens *would* be available (0 when it consumed).
+    [[nodiscard]] std::chrono::nanoseconds consume(std::uint64_t n) {
+        std::lock_guard lk(mu_);
+        if (rate_ == 0)
+            return {};
+        refill_locked();
+        if (tokens_ >= n) {
+            tokens_ -= n;
+            return {};
+        }
+        const std::uint64_t deficit = n - tokens_;
+        // ns to earn `deficit` tokens at rate_ bytes/s
+        return std::chrono::nanoseconds{
+            static_cast<std::int64_t>((deficit * 1'000'000'000ull + rate_ - 1) / rate_)};
+    }
+
+    // Two-phase for the hierarchy: check every level, then commit on all or none.
+    [[nodiscard]] std::chrono::nanoseconds peek(std::uint64_t n) {
+        std::lock_guard lk(mu_);
+        if (rate_ == 0)
+            return {};
+        refill_locked();
+        if (tokens_ >= n)
+            return {};
+        const std::uint64_t deficit = n - tokens_;
+        return std::chrono::nanoseconds{
+            static_cast<std::int64_t>((deficit * 1'000'000'000ull + rate_ - 1) / rate_)};
+    }
+    void commit(std::uint64_t n) {
+        std::lock_guard lk(mu_);
+        if (rate_ == 0)
+            return;
+        tokens_ = tokens_ >= n ? tokens_ - n : 0;
+    }
+
+  private:
+    using clock = std::chrono::steady_clock;
+    void refill_locked() {
+        auto now = clock::now();
+        auto dt = std::chrono::duration_cast<std::chrono::nanoseconds>(now - last_).count();
+        if (dt <= 0)
+            return;
+        last_ = now;
+        // added = rate_ * dt / 1e9, guarding overflow for very long idle gaps
+        long double added =
+            static_cast<long double>(rate_) * static_cast<long double>(dt) / 1'000'000'000.0L;
+        std::uint64_t add =
+            added >= static_cast<long double>(cap_) ? cap_ : static_cast<std::uint64_t>(added);
+        tokens_ = tokens_ + add > cap_ ? cap_ : tokens_ + add;
+    }
+
+    mutable std::mutex mu_;
+    std::uint64_t rate_ = 0;
+    std::uint64_t cap_ = 0;
+    std::uint64_t tokens_ = 0;
+    clock::time_point last_ = clock::now();
+};
+
+// The hierarchy. All limits default to 0 (unlimited). A task with no queue is limited by
+// task + global only.
+class RateLimiter {
+  public:
+    void set_global_limit(std::uint64_t bps) { global_.set_rate(bps); }
+    [[nodiscard]] std::uint64_t global_limit() const { return global_.rate(); }
+
+    void set_queue_limit(QueueId q, std::uint64_t bps) {
+        std::lock_guard lk(mu_);
+        queues_[q].set_rate(bps);
+    }
+    void set_task_limit(TaskId t, std::uint64_t bps) {
+        std::lock_guard lk(mu_);
+        tasks_[t].set_rate(bps);
+    }
+
+    void attach_task(TaskId t, std::optional<QueueId> q) {
+        std::lock_guard lk(mu_);
+        tasks_.try_emplace(t);
+        if (q) {
+            task_queue_[t] = *q;
+            queues_.try_emplace(*q);
+        } else {
+            task_queue_.erase(t);
+        }
+    }
+    void detach_task(TaskId t) {
+        std::lock_guard lk(mu_);
+        tasks_.erase(t);
+        task_queue_.erase(t);
+    }
+
+    // Consume `n` bytes against task, queue and global. 0 => consumed everywhere. > 0 =>
+    // consumed nowhere; wait that long and retry. Held under mu_ for its whole duration
+    // so a concurrent detach_task() can't invalidate the bucket it is using.
+    [[nodiscard]] std::chrono::nanoseconds acquire(TaskId t, std::uint64_t n) {
+        std::lock_guard lk(mu_);
+        TokenBucket *tb = nullptr;
+        TokenBucket *qb = nullptr;
+        if (auto it = tasks_.find(t); it != tasks_.end())
+            tb = &it->second;
+        if (auto qit = task_queue_.find(t); qit != task_queue_.end())
+            if (auto q = queues_.find(qit->second); q != queues_.end())
+                qb = &q->second;
+
+        // peek all, then commit all or none — no level "leaks" tokens on a partial miss.
+        std::chrono::nanoseconds wait{};
+        if (tb)
+            wait = std::max(wait, tb->peek(n));
+        if (qb)
+            wait = std::max(wait, qb->peek(n));
+        wait = std::max(wait, global_.peek(n));
+        if (wait.count() > 0)
+            return wait;
+
+        if (tb)
+            tb->commit(n);
+        if (qb)
+            qb->commit(n);
+        global_.commit(n);
+        return {};
+    }
+
+  private:
+    mutable std::mutex mu_;  // guards the maps AND serialises acquire()
+    TokenBucket global_;
+    std::unordered_map<QueueId, TokenBucket> queues_;
+    std::unordered_map<TaskId, TokenBucket> tasks_;
+    std::unordered_map<TaskId, QueueId> task_queue_;
+};
+
+}  // namespace vdm::rate
+
+#endif  // VDM_RATE_TOKEN_BUCKET_HPP

core/include/vdm/task/download.hpp

@@ -34,15 +34,17 @@ namespace vdm::task {
 // --- input ---------------------------------------------------------------------------
 
 struct Checksum {
-    enum class Algo { md5, sha1, sha256 };
+    enum class Algo { md5, sha1, sha256, sha512 };  // matches the wire Checksum set
     Algo algo = Algo::sha256;
     std::string hex;  // lower-case, no separators
 };
 
 // Everything the engine needs to run ONE download. DAEMON has already run the rules
-// engine, canonicalised the path and checked it against the allowed roots, and resolved
-// the filename — `save_path` is absolute and final. `<save_path>.veloxpart` and
-// `<save_path>.veloxpart.meta` live beside it during the transfer.
+// engine, canonicalised the path, checked it against the allowed roots, resolved the
+// filename, and created the parent directory — `save_path` is absolute and final and its
+// directory exists. `<save_path>.veloxpart` and `<save_path>.veloxpart.meta` live beside
+// it during the transfer; on success the part file is renamed in place. If the directory
+// is missing at open time the task fails with Error::path_rejected.
 struct DownloadSpec {
     std::string url;
     std::vector<std::string> mirrors;  // alternative URLs for the same bytes
@@ -185,6 +187,9 @@ class DownloadHandle {
     void pause();
     void resume();
     // Idempotent, terminal. discard_partial also removes the .veloxpart[.meta] files.
+    // Always fires on_state(_, cancelled, nullopt) then on_finished(Err{Error::canceled}),
+    // in that order. `download.cancel` == cancel(false); `download.remove` == cancel(true)
+    // (plus DAEMON's own row/file cleanup).
     void cancel(bool discard_partial = false);
 
     // Only act while the task is awaiting the matching input (auto-paused for auth /

core/tests/CMakeLists.txt

@@ -31,6 +31,7 @@ vdm_add_test(veloxcore_veloxpart_test meta/veloxpart_test.cpp)
 vdm_add_test(veloxcore_segmenter_test segment/segmenter_test.cpp)
 vdm_add_test(veloxcore_budget_test segment/budget_test.cpp)
 vdm_add_test(veloxcore_engine_api_test task/api_compiles_test.cpp)
+vdm_add_test(veloxcore_token_bucket_test rate/token_bucket_test.cpp)
 
 set(_testserver ${CMAKE_SOURCE_DIR}/tools/testserver/testserver.py)
 foreach(net_it http_client probe)

core/tests/rate/token_bucket_test.cpp

@@ -0,0 +1,145 @@
+#include "vdm/rate/token_bucket.hpp"
+
+#include <atomic>
+#include <chrono>
+#include <thread>
+#include <vector>
+
+#include "vtest.hpp"
+
+using namespace vdm;
+using namespace vdm::rate;
+using namespace std::chrono_literals;
+
+namespace {
+TaskId tid(std::uint64_t v) {
+    return TaskId{v};
+}
+QueueId qid(std::uint64_t v) {
+    return QueueId{v};
+}
+}  // namespace
+
+VT_TEST(tb_unlimited_never_waits) {
+    TokenBucket b(0);
+    for (int i = 0; i < 1000; ++i)
+        VT_CHECK_EQ(b.consume(1'000'000).count(), 0);
+}
+
+VT_TEST(tb_burst_then_throttle) {
+    // 1000 B/s, default burst = 1 s = 1000 tokens.
+    TokenBucket b(1000);
+    VT_CHECK_EQ(b.consume(1000).count(), 0);  // drains the burst
+    auto w = b.consume(1000);                 // empty now: must wait ~1 s
+    VT_CHECK(w >= 900ms && w <= 1100ms);
+}
+
+VT_TEST(tb_refills_over_time) {
+    TokenBucket b(10'000, /*burst=*/10'000);
+    VT_CHECK_EQ(b.consume(10'000).count(), 0);
+    std::this_thread::sleep_for(120ms);  // ~1200 tokens back
+    auto w = b.consume(1000);
+    VT_CHECK_EQ(w.count(), 0);  // affordable from the refill
+    auto w2 = b.consume(5000);
+    VT_CHECK(w2.count() > 0);  // not that much yet
+}
+
+VT_TEST(tb_burst_caps_accumulation) {
+    TokenBucket b(1000, /*burst=*/2000);
+    std::this_thread::sleep_for(100ms);  // idle far longer than burst/rate would fill
+    std::this_thread::sleep_for(100ms);
+    VT_CHECK_EQ(b.consume(2000).count(), 0);  // at most the 2000 cap accumulated
+    VT_CHECK(b.consume(1).count() > 0);       // and no more
+}
+
+VT_TEST(tb_set_rate_zero_makes_unlimited) {
+    TokenBucket b(1000);
+    VT_CHECK_EQ(b.consume(1000).count(), 0);
+    VT_CHECK(b.consume(1000).count() > 0);
+    b.set_rate(0);
+    VT_CHECK_EQ(b.consume(1'000'000).count(), 0);
+}
+
+// --- the hierarchy --------------------------------------------------------------------
+
+VT_TEST(rl_all_unlimited_by_default) {
+    RateLimiter rl;
+    rl.attach_task(tid(1), std::nullopt);
+    for (int i = 0; i < 100; ++i)
+        VT_CHECK_EQ(rl.acquire(tid(1), 1'000'000).count(), 0);
+}
+
+VT_TEST(rl_tightest_level_binds) {
+    RateLimiter rl;
+    rl.set_global_limit(100'000);
+    rl.set_queue_limit(qid(9), 20'000);
+    rl.set_task_limit(tid(1), 50'000);
+    rl.attach_task(tid(1), qid(9));
+
+    // burst: task 50k, queue 20k, global 100k -> the queue's 20k is the ceiling
+    VT_CHECK_EQ(rl.acquire(tid(1), 20'000).count(), 0);
+    auto w = rl.acquire(tid(1), 5'000);
+    VT_CHECK(w.count() > 0);  // queue bucket is dry even though task & global aren't
+}
+
+VT_TEST(rl_no_partial_consumption_on_miss) {
+    RateLimiter rl;
+    rl.set_global_limit(1'000'000);   // plenty
+    rl.set_task_limit(tid(1), 1000);  // 1 s burst
+    rl.attach_task(tid(1), std::nullopt);
+
+    VT_CHECK_EQ(rl.acquire(tid(1), 1000).count(), 0);  // drain the task bucket
+    for (int i = 0; i < 5; ++i)
+        VT_CHECK(rl.acquire(tid(1), 1000).count() > 0);  // task bucket blocks, repeatedly
+
+    // global must NOT have been charged for any of those blocked attempts: a fresh task
+    // limited only by the global bucket can still spend nearly its whole burst (only the
+    // one *successful* 1000-byte acquire above was charged).
+    rl.attach_task(tid(2), std::nullopt);
+    VT_CHECK_EQ(rl.acquire(tid(2), 990'000).count(), 0);
+}
+
+VT_TEST(rl_detach_then_acquire_is_safe_and_unlimited) {
+    RateLimiter rl;
+    rl.set_task_limit(tid(1), 1000);
+    rl.attach_task(tid(1), std::nullopt);
+    VT_CHECK_EQ(rl.acquire(tid(1), 1000).count(), 0);
+    rl.detach_task(tid(1));
+    // unknown task -> no task/queue bucket, only global (unlimited here)
+    VT_CHECK_EQ(rl.acquire(tid(1), 1'000'000).count(), 0);
+}
+
+VT_TEST(rl_enforces_aggregate_rate_under_load) {
+    RateLimiter rl;
+    const std::uint64_t rate = 4'000'000;  // 4 MB/s global
+    rl.set_global_limit(rate);
+    for (std::uint64_t i = 1; i <= 8; ++i)
+        rl.attach_task(tid(i), std::nullopt);
+
+    std::atomic<std::uint64_t> moved{0};
+    auto t0 = std::chrono::steady_clock::now();
+    std::vector<std::jthread> ws;
+    for (std::uint64_t i = 1; i <= 8; ++i) {
+        ws.emplace_back([&, id = tid(i)] {
+            for (int k = 0; k < 400; ++k) {
+                std::uint64_t chunk = 16 * 1024;
+                for (;;) {
+                    auto w = rl.acquire(id, chunk);
+                    if (w.count() == 0)
+                        break;
+                    std::this_thread::sleep_for(
+                        std::min<std::chrono::nanoseconds>(w, std::chrono::milliseconds(20)));
+                }
+                moved.fetch_add(chunk);
+            }
+        });
+    }
+    ws.clear();  // join
+    auto secs = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();
+
+    double effective = moved.load() / secs;
+    // Allow one burst's worth of slop plus scheduling noise: effective rate should sit
+    // within ~2x of the configured limit, never wildly above.
+    VT_CHECK(effective <= rate * 2.5);
+    VT_CHECK(moved.load() == 8u * 400u * 16u * 1024u);
+}