feat(io): add REST performance instrumentation

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

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 —

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;
 };
 

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; }
 

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,