diff --git a/include/cucascade/io/rest/config.hpp b/include/cucascade/io/rest/config.hpp
index efe2119..3eba08d 100644
--- a/include/cucascade/io/rest/config.hpp
+++ b/include/cucascade/io/rest/config.hpp
@@ -87,6 +87,12 @@ struct config {
   std::chrono::milliseconds retry_jitter{50};
   bool honor_retry_after{true};
 
+  /// When set, the reactor records the per-chunk micro timings (chunk_get,
+  /// queue_wait, ttfb, h2d_observed) into its perf counters.  The retry,
+  /// terminal-failure, device-stream-sync and payload-byte counters are always
+  /// recorded, independent of this flag.
+  bool perf_instrumentation{false};
+
   /// Suffix-range window (bytes) for the parquet footer probe
   /// (@c open_hint::parquet_footer_probe): one `Range: bytes=-N` GET resolves the
   /// object size and stashes its last N bytes, so cuDF's trailer/footer reads are
diff --git a/include/cucascade/io/rest/rest_ioctx.hpp b/include/cucascade/io/rest/rest_ioctx.hpp
index 78bc89e..9701570 100644
--- a/include/cucascade/io/rest/rest_ioctx.hpp
+++ b/include/cucascade/io/rest/rest_ioctx.hpp
@@ -57,6 +57,11 @@ class rest_ioctx : public templated_ioctx<rest_reactor> {
 
   [[nodiscard]] io_context_type type() const noexcept override { return io_context_type::restful; }
 
+  /// Pool-aggregated perf counters: per-reactor snapshots with totals and
+  /// counts summed, maxes maxed, and ttfb the smallest non-zero reactor value.
+  /// Lock-free; safe to call while the pool is running.
+  [[nodiscard]] rest_perf_snapshot perf_snapshot() const noexcept;
+
   /// Stream a bucket's ListObjectsV2 pages under @p prefix to @p sink, one call
   /// per page (a page holds at most 1000 entries, so peak memory is one page
   /// regardless of bucket population).  @p sink returns false to stop early —
diff --git a/include/cucascade/io/rest/rest_reactor.hpp b/include/cucascade/io/rest/rest_reactor.hpp
index 64c5f4c..baa319e 100644
--- a/include/cucascade/io/rest/rest_reactor.hpp
+++ b/include/cucascade/io/rest/rest_reactor.hpp
@@ -28,6 +28,7 @@
 
 #include <rmm/cuda_stream_view.hpp>
 
+#include <atomic>
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
@@ -164,6 +165,50 @@ class rest_io_object : public io_object {
   shared_byte_span _stash;
 };
 
+// ---------------------------------------------------------------------------
+// rest_perf_snapshot
+// ---------------------------------------------------------------------------
+
+/// Plain-value perf counters read out of a reactor, or summed across the pool
+/// by @c rest_ioctx.  The ns totals/maxes and ttfb stay 0 unless the reactor's
+/// @c perf_instrumentation is on; retry / terminal / device-stream-sync and
+/// payload-bytes counts are populated regardless.  The layout is not
+/// ABI-stable: consumers build from the same source pin, and fields are
+/// appended, never reordered or removed.
+struct rest_perf_snapshot {
+  std::uint64_t chunk_get_ns_total{0};
+  std::uint64_t chunk_get_count{0};
+  std::uint64_t chunk_get_ns_max{0};
+  std::uint64_t queue_wait_ns_total{0};
+  std::uint64_t queue_wait_count{0};
+  // ttfb = span from GET submission to completion of the reactor's first
+  // completed GET (async chunk or footer probe) — not first byte on the wire.
+  std::uint64_t ttfb_ns{0};
+  // h2d_observed_* time the copy_h2d_async call itself — the host-side async
+  // launch cost, not the copy, which completes later on the stream.
+  std::uint64_t h2d_observed_ns_total{0};
+  std::uint64_t h2d_observed_count{0};
+  std::uint64_t h2d_observed_ns_max{0};
+  std::uint64_t retries_total{0};
+  std::uint64_t terminal_failures_total{0};
+  std::uint64_t device_stream_sync_total{0};
+  // Always-on: HTTP response *body* bytes received (sink.total_received), summed
+  // over every completed curl attempt incl. retries / partial / failed bodies.
+  // Not TLS/header/TCP-frame bytes — this is the S3-scan payload byte budget.
+  std::uint64_t payload_bytes_read_total{0};
+  // perf_instrumentation-gated. Blocking host GETs remain part of chunk_get_*
+  // and are also attributed to blocking_host_get_*. Stash hits issue no GET and
+  // increment neither.
+  std::uint64_t blocking_host_get_count{0};
+  std::uint64_t blocking_host_get_wall_ns_total{0};
+  std::uint64_t blocking_host_get_wall_ns_max{0};
+};
+
+/// How @c prep_host_rx_request attributes the resulting GETs in the perf
+/// snapshot: a @c blocking read (synchronous host_read) is counted in
+/// blocking_host_get_* in addition to chunk_get_*.
+enum class host_read_attribution : std::uint8_t { async_chunk, blocking };
+
 // ---------------------------------------------------------------------------
 // rest_reactor
 // ---------------------------------------------------------------------------
@@ -237,6 +282,10 @@ class rest_reactor {
   static request_type_ptr prep_host_rx_request(const reactor_config_type& cfg,
                                                const io_object_type& file,
                                                const io_object_segment& segment);
+  static request_type_ptr prep_host_rx_request(const reactor_config_type& cfg,
+                                               const io_object_type& file,
+                                               const io_object_segment& segment,
+                                               host_read_attribution attribution);
 
   static request_type_ptr prep_host_rxv_request(const reactor_config_type& cfg,
                                                 const io_object_type& file,
@@ -291,12 +340,17 @@ class rest_reactor {
   /// body on HTTP 200.  @p canonical_query is the pre-encoded, key-sorted
   /// request query (no auth params — authorization is added via
   /// @c authorize_list).  @p prefix is only for retry-log / error text.
-  /// Control-plane op: retries are WARN-logged like every retry loop here, but
-  /// the XML body is never treated as object-read payload.
+  /// Control-plane op: retries/terminals are counted (and retries WARN-logged)
+  /// like every retry loop here, but the XML body never touches the chunk-GET /
+  /// payload byte counters.
   std::string list_page(std::string_view bucket,
                         std::string_view prefix,
                         std::string_view canonical_query);
 
+  /// Snapshot of this reactor's perf counters.  Lock-free (relaxed atomic
+  /// loads); safe to call while the reactor is running.
+  [[nodiscard]] rest_perf_snapshot perf_snapshot() const noexcept;
+
   // -- capabilities / factory ----------------------------------------------
 
   /// True iff @p path is an s3:// URL this reactor can serve.
@@ -344,6 +398,29 @@ class rest_reactor {
 
   std::stop_source _stop_source;
   blocking_concurrent_queue<std::unique_ptr<rest_chunked_rx_request>> _requests;
+
+  // Instrumentation counters, owned by the reactor (not worker_loop locals) so
+  // rest_ioctx can read them cross-thread.  Gating: see rest_perf_snapshot.
+  struct perf_counters {
+    std::atomic<std::uint64_t> chunk_get_ns_total{0};
+    std::atomic<std::uint64_t> chunk_get_count{0};
+    std::atomic<std::uint64_t> chunk_get_ns_max{0};
+    std::atomic<std::uint64_t> queue_wait_ns_total{0};
+    std::atomic<std::uint64_t> queue_wait_count{0};
+    std::atomic<std::uint64_t> ttfb_ns{0};
+    std::atomic<std::uint64_t> h2d_observed_ns_total{0};
+    std::atomic<std::uint64_t> h2d_observed_count{0};
+    std::atomic<std::uint64_t> h2d_observed_ns_max{0};
+    std::atomic<std::uint64_t> retries_total{0};
+    std::atomic<std::uint64_t> terminal_failures_total{0};
+    std::atomic<std::uint64_t> device_stream_sync_total{0};
+    std::atomic<std::uint64_t> payload_bytes_read_total{0};
+    std::atomic<std::uint64_t> blocking_host_get_count{0};
+    std::atomic<std::uint64_t> blocking_host_get_wall_ns_total{0};
+    std::atomic<std::uint64_t> blocking_host_get_wall_ns_max{0};
+  };
+  perf_counters _perf;
+
   std::jthread _worker;
 };
 
diff --git a/include/cucascade/io/rest/types.hpp b/include/cucascade/io/rest/types.hpp
index 0c4820b..8871b52 100644
--- a/include/cucascade/io/rest/types.hpp
+++ b/include/cucascade/io/rest/types.hpp
@@ -24,6 +24,7 @@
 
 #include <cuda_runtime.h>
 
+#include <chrono>
 #include <cstddef>
 #include <cstdint>
 #include <memory>
@@ -97,6 +98,15 @@ struct rest_chunked_rx_request {
   // needs_event_for_synchronization).
   bool staged_through_bounce{false};
 
+  // Marks synchronous network reads for blocking_host_get_* attribution.
+  bool perf_blocking_host_get{false};
+
+  // Stamped only when the reactor's perf_instrumentation is on: t_enqueue at
+  // queue insertion, t_submit at dequeue onto a connection; their delta is the
+  // queue_wait sample (attempt 0 only), and t_submit anchors the chunk_get span.
+  std::chrono::steady_clock::time_point t_enqueue{};
+  std::chrono::steady_clock::time_point t_submit{};
+
   /// True iff this read's bytes must be host->device copied after landing.
   [[nodiscard]] bool is_device() const noexcept { return cpy_req != nullptr; }
 
diff --git a/src/io/rest/rest_ioctx.cpp b/src/io/rest/rest_ioctx.cpp
index 47aa8d3..7e7f228 100644
--- a/src/io/rest/rest_ioctx.cpp
+++ b/src/io/rest/rest_ioctx.cpp
@@ -20,6 +20,7 @@
 #include <cucascade/io/rest/s3/sigv4.hpp>
 #include <cucascade/io/uri_parser.hpp>
 
+#include <algorithm>
 #include <cstddef>
 #include <cstdint>
 #include <stdexcept>
@@ -35,6 +36,34 @@ rest_ioctx::rest_ioctx(std::size_t n_reactors, std::shared_ptr<rest_reactor::rea
 {
 }
 
+rest_perf_snapshot rest_ioctx::perf_snapshot() const noexcept
+{
+  rest_perf_snapshot agg;
+  for (auto const& r : _reactors) {
+    auto const s = r->perf_snapshot();
+    agg.chunk_get_ns_total += s.chunk_get_ns_total;
+    agg.chunk_get_count += s.chunk_get_count;
+    agg.chunk_get_ns_max = std::max(agg.chunk_get_ns_max, s.chunk_get_ns_max);
+    agg.queue_wait_ns_total += s.queue_wait_ns_total;
+    agg.queue_wait_count += s.queue_wait_count;
+    if (s.ttfb_ns != 0 && (agg.ttfb_ns == 0 || s.ttfb_ns < agg.ttfb_ns)) {
+      agg.ttfb_ns = s.ttfb_ns;  // smallest non-zero first-GET latency across the pool
+    }
+    agg.h2d_observed_ns_total += s.h2d_observed_ns_total;
+    agg.h2d_observed_count += s.h2d_observed_count;
+    agg.h2d_observed_ns_max = std::max(agg.h2d_observed_ns_max, s.h2d_observed_ns_max);
+    agg.retries_total += s.retries_total;
+    agg.terminal_failures_total += s.terminal_failures_total;
+    agg.device_stream_sync_total += s.device_stream_sync_total;
+    agg.payload_bytes_read_total += s.payload_bytes_read_total;
+    agg.blocking_host_get_count += s.blocking_host_get_count;
+    agg.blocking_host_get_wall_ns_total += s.blocking_host_get_wall_ns_total;
+    agg.blocking_host_get_wall_ns_max =
+      std::max(agg.blocking_host_get_wall_ns_max, s.blocking_host_get_wall_ns_max);
+  }
+  return agg;
+}
+
 void rest_ioctx::list_objects_paged(
   std::string_view bucket,
   std::string_view prefix,
diff --git a/src/io/rest/rest_reactor.cpp b/src/io/rest/rest_reactor.cpp
index 05e4e3d..08a4463 100644
--- a/src/io/rest/rest_reactor.cpp
+++ b/src/io/rest/rest_reactor.cpp
@@ -30,6 +30,7 @@
 
 #include <algorithm>
 #include <array>
+#include <atomic>
 #include <cctype>
 #include <cerrno>
 #include <chrono>
@@ -49,6 +50,14 @@ namespace cucascade::io::rest {
 
 namespace {
 
+/// Raise @p a to @p v when @p v is larger.  Relaxed throughout: the *_ns_max
+/// counters are perf metrics and tolerate reordered updates.
+void atomic_max_relaxed(std::atomic<std::uint64_t>& a, std::uint64_t v) noexcept
+{
+  std::uint64_t cur = a.load(std::memory_order_relaxed);
+  while (v > cur && !a.compare_exchange_weak(cur, v, std::memory_order_relaxed)) {}
+}
+
 // ---- libcurl callbacks -----------------------------------------------------
 
 /// Write callback: copy curl's bytes into the sink's destination buffer at the
@@ -502,6 +511,12 @@ void rest_reactor::enqueue(request_type_ptr req)
 void rest_reactor::enqueue_chunks(std::span<std::unique_ptr<rest_chunked_rx_request>> batch)
 {
   if (batch.empty()) { return; }
+  if (_config.perf_instrumentation) {
+    auto const now = std::chrono::steady_clock::now();
+    for (auto& c : batch) {
+      if (c) { c->t_enqueue = now; }
+    }
+  }
   bool const ok = _requests.enqueue_bulk(std::make_move_iterator(batch.data()), batch.size());
   if (!ok) { throw std::runtime_error("rest_reactor::enqueue_chunks: enqueue_bulk failed"); }
   interrupt();
@@ -514,6 +529,14 @@ void rest_reactor::enqueue_chunks(std::span<std::unique_ptr<rest_chunked_rx_requ
 rest_reactor::request_type_ptr rest_reactor::prep_host_rx_request(const reactor_config_type& cfg,
                                                                   const io_object_type& file,
                                                                   const io_object_segment& segment)
+{
+  return prep_host_rx_request(cfg, file, segment, host_read_attribution::async_chunk);
+}
+
+rest_reactor::request_type_ptr rest_reactor::prep_host_rx_request(const reactor_config_type& cfg,
+                                                                  const io_object_type& file,
+                                                                  const io_object_segment& segment,
+                                                                  host_read_attribution attribution)
 {
   if (segment.size == 0) { return rest_rx_request::create({}); }
   // A host read has no bounce fallback (needs_bounce requires is_device()): a
@@ -556,12 +579,13 @@ rest_reactor::request_type_ptr rest_reactor::prep_host_rx_request(const reactor_
   chunks.reserve(n_chunks);
   size_t pos = 0;  // byte offset within the segment
   for (size_t c = 0; c < n_chunks; ++c) {
-    size_t const piece = base + (c < rem ? 1 : 0);
-    auto req           = std::make_unique<rest_chunked_rx_request>();
-    req->object        = obj;
-    req->chunk         = io_object_segment{segment.offset + pos, piece, dst + pos};
-    req->file_size     = fsize;
-    req->manager       = manager;
+    size_t const piece          = base + (c < rem ? 1 : 0);
+    auto req                    = std::make_unique<rest_chunked_rx_request>();
+    req->object                 = obj;
+    req->chunk                  = io_object_segment{segment.offset + pos, piece, dst + pos};
+    req->file_size              = fsize;
+    req->manager                = manager;
+    req->perf_blocking_host_get = (attribution == host_read_attribution::blocking);
     chunks.push_back(std::move(req));
     pos += piece;
   }
@@ -791,12 +815,37 @@ size_t rest_reactor::host_read(const io_object_type& file, size_t offset, size_t
   // full TCP+TLS handshake per call and duplicates the retry logic.  Build the
   // request, grab its future BEFORE enqueue (which moves the chunks out), then
   // block: get() rethrows the first reported error or returns the byte count.
-  auto req = prep_host_rx_request(_config, file, io_object_segment{offset, size, dst});
+  auto req = prep_host_rx_request(
+    _config, file, io_object_segment{offset, size, dst}, host_read_attribution::blocking);
   auto fut = req->get_future();
   enqueue(std::move(req));
   return std::move(fut).get();
 }
 
+rest_perf_snapshot rest_reactor::perf_snapshot() const noexcept
+{
+  rest_perf_snapshot s;
+  s.chunk_get_ns_total       = _perf.chunk_get_ns_total.load(std::memory_order_relaxed);
+  s.chunk_get_count          = _perf.chunk_get_count.load(std::memory_order_relaxed);
+  s.chunk_get_ns_max         = _perf.chunk_get_ns_max.load(std::memory_order_relaxed);
+  s.queue_wait_ns_total      = _perf.queue_wait_ns_total.load(std::memory_order_relaxed);
+  s.queue_wait_count         = _perf.queue_wait_count.load(std::memory_order_relaxed);
+  s.ttfb_ns                  = _perf.ttfb_ns.load(std::memory_order_relaxed);
+  s.h2d_observed_ns_total    = _perf.h2d_observed_ns_total.load(std::memory_order_relaxed);
+  s.h2d_observed_count       = _perf.h2d_observed_count.load(std::memory_order_relaxed);
+  s.h2d_observed_ns_max      = _perf.h2d_observed_ns_max.load(std::memory_order_relaxed);
+  s.retries_total            = _perf.retries_total.load(std::memory_order_relaxed);
+  s.terminal_failures_total  = _perf.terminal_failures_total.load(std::memory_order_relaxed);
+  s.device_stream_sync_total = _perf.device_stream_sync_total.load(std::memory_order_relaxed);
+  s.payload_bytes_read_total = _perf.payload_bytes_read_total.load(std::memory_order_relaxed);
+  s.blocking_host_get_count  = _perf.blocking_host_get_count.load(std::memory_order_relaxed);
+  s.blocking_host_get_wall_ns_total =
+    _perf.blocking_host_get_wall_ns_total.load(std::memory_order_relaxed);
+  s.blocking_host_get_wall_ns_max =
+    _perf.blocking_host_get_wall_ns_max.load(std::memory_order_relaxed);
+  return s;
+}
+
 size_t rest_reactor::head_object_size(std::string_view bucket, std::string_view key)
 {
   object_ref const obj{std::string(bucket), std::string(key)};
@@ -827,6 +876,7 @@ size_t rest_reactor::head_object_size(std::string_view bucket, std::string_view
       curl_off_t cl = -1;
       curl_easy_getinfo(h.get(), CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &cl);
       if (cl < 0) {
+        _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
         throw std::runtime_error("rest_reactor::head_object_size: missing Content-Length for " +
                                  obj.bucket + "/" + obj.key);
       }
@@ -838,10 +888,12 @@ size_t rest_reactor::head_object_size(std::string_view bucket, std::string_view
     bool const retriable =
       (rc != CURLE_OK && is_retriable_curl(rc)) || (rc == CURLE_OK && is_retriable_status(status));
     if (!retriable) {
+      _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
       throw std::runtime_error("rest_reactor::head_object_size: " + last_error + " for " +
                                obj.bucket + "/" + obj.key);
     }
     if (attempt + 1 < _config.max_retry_attempts) {
+      _perf.retries_total.fetch_add(1, std::memory_order_relaxed);
       CUCASCADE_LOG_WARN("rest_reactor::head_object_size: retrying {}/{} after {} (attempt {}/{})",
                          obj.bucket,
                          obj.key,
@@ -851,6 +903,7 @@ size_t rest_reactor::head_object_size(std::string_view bucket, std::string_view
       std::this_thread::sleep_for(compute_backoff(attempt, hc.retry_after, _config));
     }
   }
+  _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
   throw std::runtime_error("rest_reactor::head_object_size: exhausted retries (" + last_error +
                            ") for " + obj.bucket + "/" + obj.key);
 }
@@ -886,6 +939,9 @@ std::string rest_reactor::list_page(std::string_view bucket,
     long status       = 0;
     curl_easy_getinfo(h.get(), CURLINFO_RESPONSE_CODE, &status);
 
+    // Control-plane response: the XML body is deliberately NOT credited to
+    // chunk_get_count / payload_bytes_read_total — those track object-read
+    // payload, and LIST is metadata.
     if (rc == CURLE_OK && status == 200) { return body; }
 
     last_error =
@@ -893,10 +949,12 @@ std::string rest_reactor::list_page(std::string_view bucket,
     bool const retriable =
       (rc != CURLE_OK && is_retriable_curl(rc)) || (rc == CURLE_OK && is_retriable_status(status));
     if (!retriable) {
+      _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
       throw std::runtime_error("rest_reactor::list_page: " + last_error + " for " + bucket_s + "/" +
                                prefix_s);
     }
     if (attempt + 1 < _config.max_retry_attempts) {
+      _perf.retries_total.fetch_add(1, std::memory_order_relaxed);
       CUCASCADE_LOG_WARN("rest_reactor::list_page: retrying {}/{} after {} (attempt {}/{})",
                          bucket_s,
                          prefix_s,
@@ -906,6 +964,7 @@ std::string rest_reactor::list_page(std::string_view bucket,
       std::this_thread::sleep_for(compute_backoff(attempt, hc.retry_after, _config));
     }
   }
+  _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
   throw std::runtime_error("rest_reactor::list_page: exhausted retries (" + last_error + ") for " +
                            bucket_s + "/" + prefix_s);
 }
@@ -942,16 +1001,23 @@ footer_probe rest_reactor::fetch_footer_suffix(std::string_view bucket,
     CUCASCADE_CURL_CHECK(curl_easy_setopt(h.get(), CURLOPT_HEADERFUNCTION, &suffix_header_cb));
     CUCASCADE_CURL_CHECK(curl_easy_setopt(h.get(), CURLOPT_HEADERDATA, &sink));
 
+    auto const t0     = std::chrono::steady_clock::now();
     CURLcode const rc = curl_easy_perform(h.get());
     long status       = 0;
     curl_easy_getinfo(h.get(), CURLINFO_RESPONSE_CODE, &status);
 
+    // Mirror the async worker's finish(): payload bytes are always-on and
+    // per-attempt, so credit this attempt's body bytes outside the
+    // perf_instrumentation gate.
+    _perf.payload_bytes_read_total.fetch_add(sink.total_received, std::memory_order_relaxed);
+
     // suffix_write_cb aborts any non-206 body, so a CURLE_WRITE_ERROR here is our
     // own doing and the HTTP status is still valid; only a different curl error
     // (no HTTP status) is a genuine transport failure.
     if (rc != CURLE_OK && rc != CURLE_WRITE_ERROR) {
       last_error = std::string(curl_easy_strerror(rc));
       if (!is_retriable_curl(rc)) {
+        _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
         throw std::runtime_error("rest_reactor::fetch_footer_suffix: " + last_error + " for " +
                                  obj.bucket + "/" + obj.key);
       }
@@ -962,6 +1028,19 @@ footer_probe rest_reactor::fetch_footer_suffix(std::string_view bucket,
       auto const total = content_range_total(sink.content_range);
       auto const start = content_range_start(sink.content_range);
       if (total && start && *start <= *total && sink.data.size() == *total - *start) {
+        // Account the footer-suffix GET like an async chunk GET so bind-time
+        // footer reads stay visible in the perf snapshot.
+        if (_config.perf_instrumentation) {
+          auto const get_ns =
+            static_cast<std::uint64_t>(std::chrono::duration_cast<std::chrono::nanoseconds>(
+                                         std::chrono::steady_clock::now() - t0)
+                                         .count());
+          _perf.chunk_get_ns_total.fetch_add(get_ns, std::memory_order_relaxed);
+          _perf.chunk_get_count.fetch_add(1, std::memory_order_relaxed);
+          atomic_max_relaxed(_perf.chunk_get_ns_max, get_ns);
+          std::uint64_t expected = 0;
+          _perf.ttfb_ns.compare_exchange_strong(expected, get_ns, std::memory_order_relaxed);
+        }
         probe.object_size = *total;
         probe.window_lo   = *start;
         probe.bytes       = make_shared_byte_span(std::move(sink.data));
@@ -975,11 +1054,13 @@ footer_probe rest_reactor::fetch_footer_suffix(std::string_view bucket,
       last_error = "HTTP " + std::to_string(status);
     } else {
       // 404 / 403 / 401 / ... — an error a HEAD would not recover from either.
+      _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
       throw std::runtime_error("rest_reactor::fetch_footer_suffix: HTTP " + std::to_string(status) +
                                " for " + obj.bucket + "/" + obj.key);
     }
 
     if (attempt + 1 < _config.max_retry_attempts) {
+      _perf.retries_total.fetch_add(1, std::memory_order_relaxed);
       CUCASCADE_LOG_WARN(
         "rest_reactor::fetch_footer_suffix: retrying {}/{} after {} (attempt {}/{})",
         obj.bucket,
@@ -990,6 +1071,7 @@ footer_probe rest_reactor::fetch_footer_suffix(std::string_view bucket,
       std::this_thread::sleep_for(compute_backoff(attempt, sink.retry_after, _config));
     }
   }
+  _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
   throw std::runtime_error("rest_reactor::fetch_footer_suffix: exhausted retries (" + last_error +
                            ") for " + obj.bucket + "/" + obj.key);
 }
@@ -1313,6 +1395,7 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
         if (st == query_status::success) {
           it->manager->chunk_complete(it->bytes);
         } else {
+          _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
           it->manager->report_error(
             std::make_exception_ptr(std::runtime_error("rest_reactor: device H2D copy failed")));
         }
@@ -1347,6 +1430,7 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
       std::size_t const max_attempts =
         is_auth ? _config.max_auth_retry_attempts : _config.max_retry_attempts;
       if (counter + 1 >= max_attempts) {
+        _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
         req->manager->report_error(std::make_exception_ptr(std::runtime_error(
           "rest_reactor: exhausted retries for " + req->object.bucket + "/" + req->object.key)));
         return;
@@ -1354,6 +1438,7 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
       // Backoff tracks the transient-attempt count; an auth retry re-presigns
       // and reuses the current step without inflating it.
       auto const delay = compute_backoff(req->attempt, retry_after, _config);
+      _perf.retries_total.fetch_add(1, std::memory_order_relaxed);
       CUCASCADE_LOG_WARN("rest_reactor: retrying {}/{} after {} (attempt {}/{})",
                          req->object.bucket,
                          req->object.key,
@@ -1420,6 +1505,7 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
           dr->is_device() && (!dr->chunk.is_buffer_allocated() || dr->staged_through_bounce);
         if (needs_bounce && s.bounce == nullptr) {
           // Device staging requested but no host memory resource was configured.
+          _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
           dr->manager->report_error(std::make_exception_ptr(std::runtime_error(
             "rest_reactor: device staging unavailable (no host memory resource)")));
           dr.reset();
@@ -1427,6 +1513,16 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
         }
 
         s.req = std::move(dr);
+        if (_config.perf_instrumentation) {
+          auto const now  = std::chrono::steady_clock::now();
+          s.req->t_submit = now;
+          if (s.req->attempt == 0) {
+            auto const wait_ns = static_cast<std::uint64_t>(
+              std::chrono::duration_cast<std::chrono::nanoseconds>(now - s.req->t_enqueue).count());
+            _perf.queue_wait_ns_total.fetch_add(wait_ns, std::memory_order_relaxed);
+            _perf.queue_wait_count.fetch_add(1, std::memory_order_relaxed);
+          }
+        }
         if (needs_bounce) {
           s.req->chunk.set_data(s.bounce);
           // Record bounce-staging before finish() reads it: set_data has just
@@ -1446,7 +1542,11 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
     // `copying` (held until its H2D copy finishes); otherwise the caller
     // recycles the slot immediately.
     auto finish = [&](int i, CURLcode rc, long status) -> bool {
-      io_slot& s          = slots[static_cast<size_t>(i)];
+      io_slot& s = slots[static_cast<size_t>(i)];
+      // Always-on: credit this attempt's body bytes BEFORE any success / retry /
+      // terminal branching, so retried / partial / failed attempts stay in the
+      // byte budget.
+      _perf.payload_bytes_read_total.fetch_add(s.sink.total_received, std::memory_order_relaxed);
       auto& req           = *s.req;
       bool const ok_range = (status == 206) || (status == 200 && req.chunk.offset == 0);
       // A 206 must report, via Content-Range, that it delivered the exact range
@@ -1460,6 +1560,7 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
       if (rc == CURLE_OK && status == 206) {
         auto const start = content_range_start(s.hc.content_range);
         if (!start || *start != req.chunk.offset) {
+          _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
           req.manager->report_error(std::make_exception_ptr(std::runtime_error(
             "rest_reactor: 206 Content-Range mismatch (got '" + s.hc.content_range +
             "', requested offset " + std::to_string(req.chunk.offset) + ") for " +
@@ -1468,6 +1569,22 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
         }
       }
       if (rc == CURLE_OK && ok_range && s.sink.written >= req.chunk.size) {
+        if (_config.perf_instrumentation) {
+          auto const get_ns =
+            static_cast<std::uint64_t>(std::chrono::duration_cast<std::chrono::nanoseconds>(
+                                         std::chrono::steady_clock::now() - req.t_submit)
+                                         .count());
+          _perf.chunk_get_ns_total.fetch_add(get_ns, std::memory_order_relaxed);
+          _perf.chunk_get_count.fetch_add(1, std::memory_order_relaxed);
+          atomic_max_relaxed(_perf.chunk_get_ns_max, get_ns);
+          std::uint64_t expected = 0;
+          _perf.ttfb_ns.compare_exchange_strong(expected, get_ns, std::memory_order_relaxed);
+          if (req.perf_blocking_host_get) {
+            _perf.blocking_host_get_count.fetch_add(1, std::memory_order_relaxed);
+            _perf.blocking_host_get_wall_ns_total.fetch_add(get_ns, std::memory_order_relaxed);
+            atomic_max_relaxed(_perf.blocking_host_get_wall_ns_max, get_ns);
+          }
+        }
         if (req.is_device()) {
           // Issue the async H2D copy.  Bounce-staged reads need a CUDA event so
           // the slot (its bounce buffer) is only reused once the copy off it
@@ -1480,11 +1597,23 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
             ev  = &copy_events[req.cpy_req->device_id][static_cast<size_t>(i)];
             cev = ev->get();
           }
+          std::chrono::steady_clock::time_point h2d_start;
+          if (_config.perf_instrumentation) { h2d_start = std::chrono::steady_clock::now(); }
           cudaError_t const err = req.copy_h2d_async(cev);
           if (err != cudaSuccess) {
+            _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
             req.manager->report_error(err);
             return false;
           }
+          if (_config.perf_instrumentation) {
+            auto const h2d_ns =
+              static_cast<std::uint64_t>(std::chrono::duration_cast<std::chrono::nanoseconds>(
+                                           std::chrono::steady_clock::now() - h2d_start)
+                                           .count());
+            _perf.h2d_observed_ns_total.fetch_add(h2d_ns, std::memory_order_relaxed);
+            _perf.h2d_observed_count.fetch_add(1, std::memory_order_relaxed);
+            atomic_max_relaxed(_perf.h2d_observed_ns_max, h2d_ns);
+          }
           if (needs_event) {
             // Bounce buffer is still feeding the copy: hand the slot's token to
             // `copying` (with the event, manager and byte count) so the slot —
@@ -1507,6 +1636,7 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
       if (rc == CURLE_OK && status == 200 && req.chunk.offset != 0) {
         // Server ignored Range and returned the whole object: the bytes start
         // at offset 0, not req.offset — non-retriable, would loop forever.
+        _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
         req.manager->report_error(std::make_exception_ptr(
           std::runtime_error("rest_reactor: server ignored Range (HTTP 200) for " +
                              req.object.bucket + "/" + req.object.key)));
@@ -1535,6 +1665,7 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
       std::string const msg = rc != CURLE_OK
                                 ? std::string(curl_easy_strerror(rc))
                                 : (ok_range ? "short read" : "HTTP " + std::to_string(status));
+      _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed);
       req.manager->report_error(std::make_exception_ptr(std::runtime_error(
         "rest_reactor: " + msg + " for " + req.object.bucket + "/" + req.object.key)));
       return false;
@@ -1623,6 +1754,7 @@ void rest_reactor::worker_loop(const std::stop_token& stop_token)
     // the request_manager would never reach total_chunks.
     for (auto& pc : copying) {
       try {
+        _perf.device_stream_sync_total.fetch_add(1, std::memory_order_relaxed);
         pc.event->synchronize();
         pc.manager->chunk_complete(pc.bytes);
       } catch (const std::exception& e) {
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index 03de372..a1bb1d9 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -63,6 +63,7 @@ if(NOT CUCASCADE_TOPOLOGY_ONLY AND CUCASCADE_BUILD_IO)
     io/test_uri_parser.cpp
     io/cache/test_metadata_store.cpp
     io/kvikio/test_kvikio_config.cpp
+    io/rest/test_rest_perf_snapshot.cpp
     io/rest/test_shared_byte_span.cpp
     io/rest/s3/test_sigv4.cpp
     io/rest/s3/test_sigv4_authorizer.cpp
diff --git a/test/io/rest/loopback_range_server.hpp b/test/io/rest/loopback_range_server.hpp
new file mode 100644
index 0000000..a37adf8
--- /dev/null
+++ b/test/io/rest/loopback_range_server.hpp
@@ -0,0 +1,352 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cucascade/io/rest/authorizer.hpp>
+
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <unistd.h>
+
+#include <algorithm>
+#include <atomic>
+#include <cctype>
+#include <cerrno>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <stdexcept>
+#include <string>
+#include <string_view>
+#include <thread>
+#include <utility>
+#include <vector>
+
+namespace cucascade::test {
+
+struct range_fault_policy {
+  std::size_t fail_first_gets{0};
+  bool fail_all_gets{false};
+  int fail_status{503};
+  std::size_t fail_first_heads{0};
+  bool fail_all_heads{false};
+  int head_fail_status{503};
+  std::chrono::milliseconds response_delay{0};
+};
+
+struct listed_object {
+  std::string key;
+  std::uint64_t size{0};
+};
+
+class loopback_range_server {
+ public:
+  explicit loopback_range_server(std::vector<std::uint8_t> object,
+                                 range_fault_policy fault          = {},
+                                 std::vector<listed_object> listed = {})
+    : _object(std::move(object)), _fault(fault), _listed(std::move(listed))
+  {
+    if (_object.empty()) { throw std::runtime_error("loopback object must be non-empty"); }
+
+    _listen_fd = ::socket(AF_INET, SOCK_STREAM, 0);
+    if (_listen_fd < 0) { throw std::runtime_error("socket failed: " + errno_message()); }
+
+    int one = 1;
+    if (::setsockopt(_listen_fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) != 0) {
+      throw std::runtime_error("setsockopt failed: " + errno_message());
+    }
+
+    sockaddr_in addr{};
+    addr.sin_family      = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port        = 0;
+    if (::bind(_listen_fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) != 0) {
+      throw std::runtime_error("bind failed: " + errno_message());
+    }
+    if (::listen(_listen_fd, 64) != 0) {
+      throw std::runtime_error("listen failed: " + errno_message());
+    }
+
+    socklen_t len = sizeof(addr);
+    if (::getsockname(_listen_fd, reinterpret_cast<sockaddr*>(&addr), &len) != 0) {
+      throw std::runtime_error("getsockname failed: " + errno_message());
+    }
+    _port   = ntohs(addr.sin_port);
+    _thread = std::thread([this] { accept_loop(); });
+  }
+
+  ~loopback_range_server()
+  {
+    _stop.store(true, std::memory_order_relaxed);
+    if (_listen_fd >= 0) {
+      // shutdown() wakes the blocked accept(); close() alone does not reliably interrupt it.
+      ::shutdown(_listen_fd, SHUT_RDWR);
+      ::close(_listen_fd);
+      _listen_fd = -1;
+    }
+    if (_thread.joinable()) { _thread.join(); }
+    for (auto& worker : _workers) {
+      if (worker.joinable()) { worker.join(); }
+    }
+  }
+
+  loopback_range_server(loopback_range_server const&)            = delete;
+  loopback_range_server& operator=(loopback_range_server const&) = delete;
+
+  [[nodiscard]] std::string endpoint() const { return "http://127.0.0.1:" + std::to_string(_port); }
+
+  [[nodiscard]] std::size_t head_count() const noexcept { return _head_count.load(); }
+  [[nodiscard]] std::size_t get_count() const noexcept { return _get_count.load(); }
+  [[nodiscard]] std::size_t list_count() const noexcept { return _list_count.load(); }
+
+ private:
+  static std::string errno_message() { return std::strerror(errno); }
+
+  void accept_loop()
+  {
+    while (!_stop.load(std::memory_order_relaxed)) {
+      sockaddr_in client{};
+      socklen_t len = sizeof(client);
+      int fd        = ::accept(_listen_fd, reinterpret_cast<sockaddr*>(&client), &len);
+      if (fd < 0) {
+        if (_stop.load(std::memory_order_relaxed)) { return; }
+        continue;
+      }
+      std::scoped_lock lock{_workers_mutex};
+      _workers.emplace_back([this, fd] {
+        handle_client(fd);
+        ::close(fd);
+      });
+    }
+  }
+
+  void handle_client(int fd)
+  {
+    timeval timeout{};
+    timeout.tv_sec = 5;
+    (void)::setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout));
+
+    std::string request(8192, '\0');
+    ssize_t const n = ::recv(fd, request.data(), request.size(), 0);
+    if (n <= 0) { return; }
+    request.resize(static_cast<std::size_t>(n));
+
+    bool const is_head = request.rfind("HEAD ", 0) == 0;
+    bool const is_get  = request.rfind("GET ", 0) == 0;
+    bool const is_list = is_get && request_target(request).find("list-type=2") != std::string::npos;
+
+    if (_fault.response_delay.count() > 0) { std::this_thread::sleep_for(_fault.response_delay); }
+
+    if (is_head) {
+      auto const head_idx = _head_count.fetch_add(1, std::memory_order_relaxed);
+      if (_fault.fail_all_heads || head_idx < _fault.fail_first_heads) {
+        send_all(fd,
+                 "HTTP/1.1 " + std::to_string(_fault.head_fail_status) +
+                   " Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
+        return;
+      }
+      send_all(fd,
+               "HTTP/1.1 200 OK\r\nContent-Length: " + std::to_string(_object.size()) +
+                 "\r\nConnection: close\r\n\r\n");
+      return;
+    }
+
+    if (is_list) {
+      _list_count.fetch_add(1, std::memory_order_relaxed);
+      auto const body = list_xml();
+      send_all(fd,
+               "HTTP/1.1 200 OK\r\nContent-Type: application/xml\r\nContent-Length: " +
+                 std::to_string(body.size()) + "\r\nConnection: close\r\n\r\n" + body);
+      return;
+    }
+
+    if (!is_get) {
+      send_all(fd,
+               "HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
+      return;
+    }
+
+    auto const get_idx = _get_count.fetch_add(1, std::memory_order_relaxed);
+    if (_fault.fail_all_gets || get_idx < _fault.fail_first_gets) {
+      send_all(fd,
+               "HTTP/1.1 " + std::to_string(_fault.fail_status) +
+                 " Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
+      return;
+    }
+
+    if (auto range = parse_range(request)) {
+      auto const [start, end] = *range;
+      auto const size         = end - start + 1;
+      send_all(fd,
+               "HTTP/1.1 206 Partial Content\r\nContent-Length: " + std::to_string(size) +
+                 "\r\nContent-Range: bytes " + std::to_string(start) + "-" + std::to_string(end) +
+                 "/" + std::to_string(_object.size()) + "\r\nConnection: close\r\n\r\n");
+      send_all(fd, _object.data() + start, size);
+      return;
+    }
+
+    send_all(fd,
+             "HTTP/1.1 200 OK\r\nContent-Length: " + std::to_string(_object.size()) +
+               "\r\nConnection: close\r\n\r\n");
+    send_all(fd, _object.data(), _object.size());
+  }
+
+  static std::string request_target(std::string const& request)
+  {
+    auto const first = request.find(' ');
+    if (first == std::string::npos) { return {}; }
+    auto const second = request.find(' ', first + 1);
+    if (second == std::string::npos) { return {}; }
+    return request.substr(first + 1, second - first - 1);
+  }
+
+  static std::string xml_escape(std::string_view value)
+  {
+    std::string out;
+    for (char c : value) {
+      switch (c) {
+        case '&': out += "&amp;"; break;
+        case '<': out += "&lt;"; break;
+        case '>': out += "&gt;"; break;
+        case '\"': out += "&quot;"; break;
+        case '\'': out += "&apos;"; break;
+        default: out.push_back(c); break;
+      }
+    }
+    return out;
+  }
+
+  [[nodiscard]] std::string list_xml() const
+  {
+    std::string body =
+      "<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
+      "<ListBucketResult><IsTruncated>false</IsTruncated>";
+    for (auto const& object : _listed) {
+      body += "<Contents><Key>" + xml_escape(object.key) + "</Key><Size>" +
+              std::to_string(object.size) + "</Size></Contents>";
+    }
+    body += "</ListBucketResult>";
+    return body;
+  }
+
+  [[nodiscard]] std::optional<std::pair<std::size_t, std::size_t>> parse_range(
+    std::string const& request) const
+  {
+    std::string lower = request;
+    std::transform(lower.begin(), lower.end(), lower.begin(), [](unsigned char c) {
+      return static_cast<char>(std::tolower(c));
+    });
+    std::string const prefix{"range: bytes="};
+    auto pos = lower.find(prefix);
+    if (pos == std::string::npos) { return std::nullopt; }
+    pos += prefix.size();
+    auto const eol     = lower.find("\r\n", pos);
+    auto const end_pos = eol == std::string::npos ? lower.size() : eol;
+    auto const spec    = lower.substr(pos, end_pos - pos);
+    auto const dash    = spec.find('-');
+    if (dash == std::string::npos) { return std::nullopt; }
+
+    try {
+      std::size_t start = 0;
+      std::size_t end   = _object.size() - 1;
+      if (dash == 0) {
+        auto const suffix = static_cast<std::size_t>(std::stoull(spec.substr(1)));
+        if (suffix == 0) { return std::nullopt; }
+        start = suffix >= _object.size() ? 0 : _object.size() - suffix;
+      } else {
+        start = static_cast<std::size_t>(std::stoull(spec.substr(0, dash)));
+        if (dash + 1 < spec.size()) {
+          end = static_cast<std::size_t>(std::stoull(spec.substr(dash + 1)));
+        }
+      }
+      if (start >= _object.size()) { return std::nullopt; }
+      end = std::min(end, _object.size() - 1);
+      if (end < start) { return std::nullopt; }
+      return std::pair{start, end};
+    } catch (...) {
+      return std::nullopt;
+    }
+  }
+
+  static void send_all(int fd, std::string_view bytes)
+  {
+    send_all(fd, reinterpret_cast<std::uint8_t const*>(bytes.data()), bytes.size());
+  }
+
+  static void send_all(int fd, std::uint8_t const* bytes, std::size_t size)
+  {
+    std::size_t sent = 0;
+    // Fault tests can disconnect mid-response; suppress SIGPIPE so send() reports the failure.
+    while (sent < size) {
+      ssize_t const n = ::send(fd, bytes + sent, size - sent, MSG_NOSIGNAL);
+      if (n <= 0) { return; }
+      sent += static_cast<std::size_t>(n);
+    }
+  }
+
+  int _listen_fd{-1};
+  std::uint16_t _port{0};
+  std::vector<std::uint8_t> _object;
+  range_fault_policy _fault;
+  std::vector<listed_object> _listed;
+  std::atomic<bool> _stop{false};
+  std::atomic<std::size_t> _head_count{0};
+  std::atomic<std::size_t> _get_count{0};
+  std::atomic<std::size_t> _list_count{0};
+  std::thread _thread;
+  std::mutex _workers_mutex;
+  std::vector<std::thread> _workers;
+};
+
+class list_capable_mock_authorizer final : public io::rest::request_authorizer {
+ public:
+  explicit list_capable_mock_authorizer(std::string endpoint) : _endpoint(std::move(endpoint)) {}
+
+  io::rest::authorized_request authorize(io::rest::object_ref const& obj,
+                                         io::rest::request_method,
+                                         std::chrono::seconds) override
+  {
+    _object_calls.fetch_add(1, std::memory_order_relaxed);
+    return {_endpoint + "/" + obj.bucket + "/" + obj.key, {}};
+  }
+
+  io::rest::authorized_request authorize_list(std::string_view bucket,
+                                              std::string_view canonical_query,
+                                              std::chrono::seconds) override
+  {
+    _list_calls.fetch_add(1, std::memory_order_relaxed);
+    return {_endpoint + "/" + std::string{bucket} + "?" + std::string{canonical_query}, {}};
+  }
+
+  [[nodiscard]] int object_calls() const noexcept { return _object_calls.load(); }
+  [[nodiscard]] int list_calls() const noexcept { return _list_calls.load(); }
+
+ private:
+  std::string _endpoint;
+  std::atomic<int> _object_calls{0};
+  std::atomic<int> _list_calls{0};
+};
+
+}  // namespace cucascade::test
diff --git a/test/io/rest/test_rest_perf_snapshot.cpp b/test/io/rest/test_rest_perf_snapshot.cpp
new file mode 100644
index 0000000..5b2ac5b
--- /dev/null
+++ b/test/io/rest/test_rest_perf_snapshot.cpp
@@ -0,0 +1,594 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "loopback_range_server.hpp"
+
+#include <cucascade/io/rest/mock_authorizer.hpp>
+#include <cucascade/io/rest/rest_ioctx.hpp>
+#include <cucascade/memory/fixed_size_host_memory_resource.hpp>
+
+#include <rmm/cuda_stream.hpp>
+#include <rmm/mr/pinned_host_memory_resource.hpp>
+
+#include <cuda_runtime.h>
+
+#include <catch2/catch_all.hpp>
+
+#include <algorithm>
+#include <array>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <span>
+#include <stdexcept>
+#include <string>
+#include <type_traits>
+#include <utility>
+#include <vector>
+
+namespace {
+
+using cucascade::io::open_hint;
+using cucascade::io::rest::config;
+using cucascade::io::rest::mock_authorizer;
+using cucascade::io::rest::rest_ioctx;
+using cucascade::io::rest::rest_perf_snapshot;
+using cucascade::io::rest::rest_reactor;
+using cucascade::test::list_capable_mock_authorizer;
+using cucascade::test::listed_object;
+using cucascade::test::loopback_range_server;
+using cucascade::test::range_fault_policy;
+using namespace std::chrono_literals;
+
+std::vector<std::uint8_t> deterministic_payload(std::size_t size)
+{
+  std::vector<std::uint8_t> bytes(size);
+  for (std::size_t i = 0; i < bytes.size(); ++i) {
+    bytes[i] = static_cast<std::uint8_t>((i * 131U + 17U) & 0xffU);
+  }
+  return bytes;
+}
+
+void require_bytes_equal(std::span<const std::uint8_t> actual,
+                         std::span<const std::uint8_t> expected)
+{
+  REQUIRE(actual.size() == expected.size());
+  CHECK(std::equal(actual.begin(), actual.end(), expected.begin(), expected.end()));
+}
+
+config test_config(bool instrumentation = true)
+{
+  config cfg{};
+  cfg.request_timeout_s       = 5;
+  cfg.tls_verify              = false;
+  cfg.max_connections         = 2;
+  cfg.chunk_size              = 64 * 1024;
+  cfg.max_read_split          = 1;
+  cfg.max_retry_attempts      = 3;
+  cfg.max_auth_retry_attempts = 2;
+  cfg.retry_backoff_base      = 1ms;
+  cfg.retry_jitter            = 0ms;
+  cfg.honor_retry_after       = false;
+  cfg.perf_instrumentation    = instrumentation;
+  cfg.footer_probe_bytes      = 512;
+  return cfg;
+}
+
+struct direct_ioctx {
+  std::shared_ptr<mock_authorizer> authorizer;
+  std::shared_ptr<rest_ioctx> ioctx;
+};
+
+direct_ioctx make_ioctx(loopback_range_server const& server,
+                        config cfg,
+                        std::size_t reactors                                        = 1,
+                        cucascade::memory::fixed_size_host_memory_resource* host_mr = nullptr)
+{
+  auto authorizer = std::make_shared<mock_authorizer>(
+    cucascade::io::rest::authorized_request{server.endpoint() + "/bucket/object.bin", {}});
+  auto context = std::make_shared<rest_reactor::reactor_context>(cfg, authorizer, host_mr);
+  auto ioctx   = std::make_shared<rest_ioctx>(reactors, std::move(context));
+  ioctx->start();
+  return {std::move(authorizer), std::move(ioctx)};
+}
+
+std::unique_ptr<rest_reactor> make_reactor(
+  loopback_range_server const& server,
+  config cfg,
+  std::shared_ptr<mock_authorizer>* authorizer_out = nullptr)
+{
+  auto authorizer = std::make_shared<mock_authorizer>(
+    cucascade::io::rest::authorized_request{server.endpoint() + "/bucket/object.bin", {}});
+  auto context = std::make_shared<rest_reactor::reactor_context>(cfg, authorizer, nullptr);
+  if (authorizer_out != nullptr) { *authorizer_out = authorizer; }
+  return std::make_unique<rest_reactor>(std::move(context), "rest-perf-test");
+}
+
+std::shared_ptr<cucascade::io::io_object> known_object(rest_ioctx& ioctx, std::size_t size)
+{
+  return ioctx.open_io_object("s3://bucket/object.bin", static_cast<std::uint64_t>(size));
+}
+
+void check_micro_counters_zero(rest_perf_snapshot const& snapshot)
+{
+  CHECK(snapshot.chunk_get_ns_total == 0);
+  CHECK(snapshot.chunk_get_count == 0);
+  CHECK(snapshot.chunk_get_ns_max == 0);
+  CHECK(snapshot.queue_wait_ns_total == 0);
+  CHECK(snapshot.queue_wait_count == 0);
+  CHECK(snapshot.ttfb_ns == 0);
+  CHECK(snapshot.h2d_observed_ns_total == 0);
+  CHECK(snapshot.h2d_observed_count == 0);
+  CHECK(snapshot.h2d_observed_ns_max == 0);
+  CHECK(snapshot.blocking_host_get_count == 0);
+  CHECK(snapshot.blocking_host_get_wall_ns_total == 0);
+  CHECK(snapshot.blocking_host_get_wall_ns_max == 0);
+}
+
+class device_allocation {
+ public:
+  explicit device_allocation(std::size_t size)
+  {
+    if (cudaMalloc(reinterpret_cast<void**>(&_data), size) != cudaSuccess) {
+      throw std::runtime_error("cudaMalloc failed");
+    }
+  }
+  ~device_allocation()
+  {
+    if (_data != nullptr) { (void)cudaFree(_data); }
+  }
+
+  device_allocation(device_allocation const&)            = delete;
+  device_allocation& operator=(device_allocation const&) = delete;
+
+  [[nodiscard]] std::uint8_t* data() noexcept { return _data; }
+
+ private:
+  std::uint8_t* _data{nullptr};
+};
+
+struct staging_resource {
+  static constexpr std::size_t block_size = 64 * 1024;
+  static constexpr std::size_t capacity   = 4 * 1024 * 1024;
+
+  rmm::mr::pinned_host_memory_resource pinned;
+  cucascade::memory::fixed_size_host_memory_resource blocks{
+    0, pinned, capacity, capacity, block_size, 4, 1};
+};
+
+}  // namespace
+
+TEST_CASE("default perf snapshot is zero", "[rest][perf]")
+{
+  rest_perf_snapshot snapshot{};
+  check_micro_counters_zero(snapshot);
+  CHECK(snapshot.retries_total == 0);
+  CHECK(snapshot.terminal_failures_total == 0);
+  CHECK(snapshot.device_stream_sync_total == 0);
+  CHECK(snapshot.payload_bytes_read_total == 0);
+}
+
+TEST_CASE("perf instrumentation defaults off", "[rest][perf]")
+{
+  CHECK_FALSE(config{}.perf_instrumentation);
+}
+
+TEST_CASE("perf snapshot readouts are noexcept", "[rest][perf]")
+{
+  STATIC_REQUIRE(noexcept(std::declval<rest_reactor const&>().perf_snapshot()));
+  STATIC_REQUIRE(noexcept(std::declval<rest_ioctx const&>().perf_snapshot()));
+}
+
+TEST_CASE("ranged get feeds chunk counters", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  loopback_range_server server(payload);
+  auto fixture = make_ioctx(server, test_config());
+  auto object  = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 257> out{};
+
+  auto future = fixture.ioctx->host_read_async_io(*object, 17, out.size(), out.data());
+  REQUIRE(std::move(future).get(5s) == out.size());
+  require_bytes_equal(out, std::span<const std::uint8_t>(payload.data() + 17, out.size()));
+
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.chunk_get_count == 1);
+  CHECK(snapshot.chunk_get_ns_total > 0);
+  CHECK(snapshot.chunk_get_ns_max > 0);
+  CHECK(snapshot.payload_bytes_read_total == out.size());
+  CHECK(server.get_count() == 1);
+}
+
+TEST_CASE("head request feeds retry and terminal counters", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+
+  SECTION("clean head leaves data counters unchanged")
+  {
+    loopback_range_server server(payload);
+    auto reactor = make_reactor(server, test_config());
+    CHECK(reactor->head_object_size("bucket", "object.bin") == payload.size());
+    auto const snapshot = reactor->perf_snapshot();
+    CHECK(snapshot.chunk_get_count == 0);
+    CHECK(snapshot.payload_bytes_read_total == 0);
+    CHECK(snapshot.retries_total == 0);
+    CHECK(snapshot.terminal_failures_total == 0);
+    CHECK(server.head_count() == 1);
+  }
+
+  SECTION("transient head failure is retried")
+  {
+    range_fault_policy fault;
+    fault.fail_first_heads = 1;
+    loopback_range_server server(payload, fault);
+    auto reactor = make_reactor(server, test_config(false));
+    CHECK(reactor->head_object_size("bucket", "object.bin") == payload.size());
+    auto const snapshot = reactor->perf_snapshot();
+    CHECK(snapshot.retries_total == 1);
+    CHECK(snapshot.terminal_failures_total == 0);
+    CHECK(server.head_count() == 2);
+  }
+}
+
+TEST_CASE("footer probe attributes as chunk get", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  loopback_range_server server(payload);
+  auto reactor = make_reactor(server, test_config());
+
+  auto const probe = reactor->fetch_footer_suffix("bucket", "object.bin", 512);
+  REQUIRE(probe.bytes != nullptr);
+  CHECK(probe.object_size == payload.size());
+  CHECK(probe.window_lo == payload.size() - 512);
+
+  auto const snapshot = reactor->perf_snapshot();
+  CHECK(snapshot.chunk_get_count == 1);
+  CHECK(snapshot.blocking_host_get_count == 0);
+  CHECK(snapshot.payload_bytes_read_total == 512);
+  CHECK(server.get_count() == 1);
+}
+
+TEST_CASE("blocking host read is counted additively", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  loopback_range_server server(payload);
+  auto fixture = make_ioctx(server, test_config());
+  auto object  = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 128> out{};
+
+  REQUIRE(fixture.ioctx->host_read_io(*object, 99, out.size(), out.data()) == out.size());
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.chunk_get_count == 1);
+  CHECK(snapshot.blocking_host_get_count == 1);
+  CHECK(snapshot.blocking_host_get_wall_ns_total > 0);
+  CHECK(snapshot.blocking_host_get_wall_ns_max > 0);
+}
+
+TEST_CASE("list bytes stay out of payload counters", "[rest][perf]")
+{
+  auto payload = deterministic_payload(128);
+  loopback_range_server server(payload, {}, {listed_object{"prefix/a.parquet", 17}});
+  auto authorizer = std::make_shared<list_capable_mock_authorizer>(server.endpoint());
+  auto context =
+    std::make_shared<rest_reactor::reactor_context>(test_config(), authorizer, nullptr);
+  auto ioctx = std::make_shared<rest_ioctx>(1, std::move(context));
+
+  auto const listed = ioctx->list_objects("bucket", "prefix/");
+  REQUIRE(listed.size() == 1);
+  CHECK(listed[0].key == "prefix/a.parquet");
+  CHECK(listed[0].size == 17);
+  auto const snapshot = ioctx->perf_snapshot();
+  CHECK(snapshot.payload_bytes_read_total == 0);
+  CHECK(snapshot.chunk_get_count == 0);
+  CHECK(authorizer->list_calls() == 1);
+  CHECK(server.list_count() == 1);
+}
+
+TEST_CASE("gate off keeps safety counters live", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+
+  SECTION("retry and payload counters remain active")
+  {
+    range_fault_policy fault;
+    fault.fail_first_gets = 1;
+    loopback_range_server server(payload, fault);
+    auto fixture = make_ioctx(server, test_config(false));
+    auto object  = known_object(*fixture.ioctx, payload.size());
+    std::array<std::uint8_t, 128> out{};
+    auto future = fixture.ioctx->host_read_async_io(*object, 0, out.size(), out.data());
+    REQUIRE(std::move(future).get(5s) == out.size());
+
+    auto const snapshot = fixture.ioctx->perf_snapshot();
+    check_micro_counters_zero(snapshot);
+    CHECK(snapshot.payload_bytes_read_total == out.size());
+    CHECK(snapshot.retries_total == 1);
+    CHECK(snapshot.terminal_failures_total == 0);
+    CHECK(snapshot.device_stream_sync_total == 0);
+  }
+
+  SECTION("terminal counters remain active")
+  {
+    range_fault_policy fault;
+    fault.fail_all_gets = true;
+    fault.fail_status   = 404;
+    loopback_range_server server(payload, fault);
+    auto fixture = make_ioctx(server, test_config(false));
+    auto object  = known_object(*fixture.ioctx, payload.size());
+    std::array<std::uint8_t, 128> out{};
+    auto future = fixture.ioctx->host_read_async_io(*object, 0, out.size(), out.data());
+    CHECK_THROWS(std::move(future).get(5s));
+
+    auto const snapshot = fixture.ioctx->perf_snapshot();
+    check_micro_counters_zero(snapshot);
+    CHECK(snapshot.retries_total == 0);
+    CHECK(snapshot.terminal_failures_total == 1);
+  }
+}
+
+TEST_CASE("gate on records micro timings", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  loopback_range_server server(payload);
+  auto fixture = make_ioctx(server, test_config());
+  auto object  = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 128> out{};
+
+  REQUIRE(fixture.ioctx->host_read_io(*object, 0, out.size(), out.data()) == out.size());
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.chunk_get_count == 1);
+  CHECK(snapshot.chunk_get_ns_total > 0);
+  CHECK(snapshot.chunk_get_ns_max > 0);
+  CHECK(snapshot.queue_wait_count == 1);
+  CHECK(snapshot.queue_wait_ns_total > 0);
+  CHECK(snapshot.ttfb_ns > 0);
+  CHECK(snapshot.blocking_host_get_count == 1);
+}
+
+TEST_CASE("pool snapshot aggregates across reactors", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  loopback_range_server server(payload);
+  auto cfg            = test_config();
+  cfg.max_connections = 1;
+  auto fixture        = make_ioctx(server, cfg, 2);
+  auto object         = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 64> first{};
+  std::array<std::uint8_t, 64> second{};
+
+  auto first_future = fixture.ioctx->host_read_async_io(*object, 0, first.size(), first.data());
+  REQUIRE(std::move(first_future).get(5s) == first.size());
+  auto second_future =
+    fixture.ioctx->host_read_async_io(*object, 128, second.size(), second.data());
+  REQUIRE(std::move(second_future).get(5s) == second.size());
+
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.chunk_get_count == 2);
+  CHECK(snapshot.chunk_get_ns_total >= snapshot.chunk_get_ns_max);
+  CHECK(snapshot.chunk_get_ns_max > 0);
+  CHECK(snapshot.queue_wait_count == 2);
+  CHECK(snapshot.ttfb_ns > 0);
+  CHECK(snapshot.payload_bytes_read_total == first.size() + second.size());
+}
+
+TEST_CASE("transient retry is counted", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  range_fault_policy fault;
+  fault.fail_first_gets = 1;
+  loopback_range_server server(payload, fault);
+  auto fixture = make_ioctx(server, test_config(false));
+  auto object  = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 128> out{};
+
+  auto future = fixture.ioctx->host_read_async_io(*object, 0, out.size(), out.data());
+  REQUIRE(std::move(future).get(5s) == out.size());
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.retries_total == 1);
+  CHECK(snapshot.terminal_failures_total == 0);
+  CHECK(fixture.authorizer->get_count() == 2);
+}
+
+TEST_CASE("exhausted retries count terminal", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  range_fault_policy fault;
+  fault.fail_all_gets = true;
+  loopback_range_server server(payload, fault);
+  auto cfg               = test_config(false);
+  cfg.max_retry_attempts = 3;
+  auto fixture           = make_ioctx(server, cfg);
+  auto object            = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 128> out{};
+
+  auto future = fixture.ioctx->host_read_async_io(*object, 0, out.size(), out.data());
+  CHECK_THROWS(std::move(future).get(5s));
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.retries_total == 2);
+  CHECK(snapshot.terminal_failures_total == 1);
+  CHECK(fixture.authorizer->get_count() == 3);
+}
+
+TEST_CASE("auth retry re-authorizes and is counted", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  range_fault_policy fault;
+  fault.fail_first_gets = 1;
+  fault.fail_status     = 403;
+  loopback_range_server server(payload, fault);
+  auto cfg                    = test_config(false);
+  cfg.max_auth_retry_attempts = 2;
+  auto fixture                = make_ioctx(server, cfg);
+  auto object                 = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 128> out{};
+
+  auto future = fixture.ioctx->host_read_async_io(*object, 0, out.size(), out.data());
+  REQUIRE(std::move(future).get(5s) == out.size());
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.retries_total == 1);
+  CHECK(snapshot.terminal_failures_total == 0);
+  CHECK(fixture.authorizer->get_count() == 2);
+}
+
+TEST_CASE("not found counts terminal", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  range_fault_policy fault;
+  fault.fail_all_gets = true;
+  fault.fail_status   = 404;
+  loopback_range_server server(payload, fault);
+  auto fixture = make_ioctx(server, test_config(false));
+  auto object  = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 128> out{};
+
+  auto future = fixture.ioctx->host_read_async_io(*object, 0, out.size(), out.data());
+  CHECK_THROWS(std::move(future).get(5s));
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.retries_total == 0);
+  CHECK(snapshot.terminal_failures_total == 1);
+  CHECK(fixture.authorizer->get_count() == 1);
+}
+
+TEST_CASE("retry events count exactly once", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  range_fault_policy fault;
+  fault.fail_all_gets = true;
+  fault.fail_status   = 503;
+  loopback_range_server server(payload, fault);
+  auto cfg               = test_config(false);
+  cfg.max_retry_attempts = 3;
+  auto fixture           = make_ioctx(server, cfg);
+  auto object            = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 64> out{};
+
+  for (int i = 0; i < 2; ++i) {
+    auto future = fixture.ioctx->host_read_async_io(*object, 0, out.size(), out.data());
+    CHECK_THROWS(std::move(future).get(5s));
+  }
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.retries_total == 4);
+  CHECK(snapshot.terminal_failures_total == 2);
+  CHECK(fixture.authorizer->get_count() == 6);
+}
+
+TEST_CASE("reactor teardown resolves all futures", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  range_fault_policy fault;
+  fault.response_delay = 500ms;
+  loopback_range_server server(payload, fault);
+  auto fixture = make_ioctx(server, test_config(false));
+  auto object  = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 128> out{};
+
+  auto future = fixture.ioctx->host_read_async_io(*object, 0, out.size(), out.data());
+  fixture.ioctx->shutdown();
+  CHECK_THROWS(std::move(future).get(2s));
+}
+
+TEST_CASE("device read records h2d timings", "[rest][perf][gpu]")
+{
+  int device_count = 0;
+  if (cudaGetDeviceCount(&device_count) != cudaSuccess || device_count == 0) {
+    SKIP("CUDA device is unavailable");
+  }
+  REQUIRE(cudaSetDevice(0) == cudaSuccess);
+
+  auto payload = deterministic_payload(32 * 1024);
+  loopback_range_server server(payload);
+  staging_resource staging;
+  auto cfg              = test_config();
+  cfg.max_connections   = 1;
+  cfg.bounce_block_size = staging_resource::block_size;
+  auto fixture          = make_ioctx(server, cfg, 1, &staging.blocks);
+  auto object           = known_object(*fixture.ioctx, payload.size());
+  device_allocation device(payload.size());
+  rmm::cuda_stream stream;
+
+  auto future =
+    fixture.ioctx->device_read_async_io(*object, 0, payload.size(), device.data(), stream.view());
+  REQUIRE(std::move(future).get(5s) == payload.size());
+  stream.synchronize();
+
+  std::vector<std::uint8_t> actual(payload.size());
+  REQUIRE(cudaMemcpy(actual.data(), device.data(), actual.size(), cudaMemcpyDeviceToHost) ==
+          cudaSuccess);
+  require_bytes_equal(actual, payload);
+
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.h2d_observed_count == 1);
+  CHECK(snapshot.h2d_observed_ns_total > 0);
+  CHECK(snapshot.h2d_observed_ns_max > 0);
+  CHECK(snapshot.device_stream_sync_total == 0);
+}
+
+TEST_CASE("queued get records queue wait", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  range_fault_policy fault;
+  fault.response_delay = 75ms;
+  loopback_range_server server(payload, fault);
+  auto cfg            = test_config();
+  cfg.max_connections = 1;
+  auto fixture        = make_ioctx(server, cfg);
+  auto object         = known_object(*fixture.ioctx, payload.size());
+  std::array<std::uint8_t, 128> first{};
+  std::array<std::uint8_t, 128> second{};
+
+  auto first_future = fixture.ioctx->host_read_async_io(*object, 0, first.size(), first.data());
+  auto second_future =
+    fixture.ioctx->host_read_async_io(*object, 256, second.size(), second.data());
+  REQUIRE(std::move(first_future).get(5s) == first.size());
+  REQUIRE(std::move(second_future).get(5s) == second.size());
+
+  auto const snapshot = fixture.ioctx->perf_snapshot();
+  CHECK(snapshot.queue_wait_count == 2);
+  // With one connection, the second GET queues behind the 75 ms delay; 50 ms leaves scheduler
+  // slack.
+  CHECK(
+    snapshot.queue_wait_ns_total >=
+    static_cast<std::uint64_t>(std::chrono::duration_cast<std::chrono::nanoseconds>(50ms).count()));
+  CHECK(snapshot.chunk_get_count == 2);
+}
+
+TEST_CASE("stash hit moves no counters", "[rest][perf]")
+{
+  auto payload = deterministic_payload(4096);
+  loopback_range_server server(payload);
+  auto cfg               = test_config();
+  cfg.footer_probe_bytes = 512;
+  auto fixture           = make_ioctx(server, cfg);
+  auto object =
+    fixture.ioctx->open_io_object("s3://bucket/object.bin", open_hint::parquet_footer_probe);
+  auto const before = fixture.ioctx->perf_snapshot();
+  std::array<std::uint8_t, 64> out{};
+
+  REQUIRE(fixture.ioctx->host_read_io(
+            *object, payload.size() - out.size(), out.size(), out.data()) == out.size());
+  require_bytes_equal(
+    out, std::span<const std::uint8_t>(payload.data() + payload.size() - out.size(), out.size()));
+  auto const after = fixture.ioctx->perf_snapshot();
+  CHECK(after.chunk_get_count == before.chunk_get_count);
+  CHECK(after.payload_bytes_read_total == before.payload_bytes_read_total);
+  CHECK(after.blocking_host_get_count == before.blocking_host_get_count);
+  CHECK(after.retries_total == before.retries_total);
+  CHECK(after.terminal_failures_total == before.terminal_failures_total);
+  CHECK(server.get_count() == 1);
+}