feat(io): S3 LIST + footer probe + kvikIO backend, and io-layer fixes (#172)

Adds three capabilities to the `io` datasource layer, fixes several latent defects found while auditing it, and gives the layer its first test coverage.

## Bug fixes

- **`io_request`: `copy_async` validates its host source before forming the pointer.** A null `host_buffer`, or an out-of-range `[src_off, src_off + size)`, previously produced UB (`nullptr + offset`) or a wild in-range pointer that the near-null check could not catch. Now returns `cudaErrorInvalidValue`. The existing asserts compile out in release, so the check has to happen before the pointer is formed.
- **`prefetching_cache`: mark the *request* state load-failed on the `prefetch_loop` cancel/stop path.** Only the chunk states were being marked, leaving the request-level state inconsistent on cancellation.
- **Four empty `catch` blocks were silently swallowing exceptions** — the `rest_reactor` and `uring_reactor` worker loops, and two in `templated_ioctx::shutdown`. Restoring the log call sites fills them. `CUCASCADE_LOG_*` are no-ops today, so this costs nothing at runtime but keeps the diagnostics in place for when logging is wired up.

## Behavioral fixes

- **`rest_reactor::preferred_prefetching_stage`: `opportunistic` → `just_in_time`.** `prefetching_cache` branches on exactly this value, so S3 reads were being eagerly prefetched into cache buffers rather than read straight into the caller's buffer. Network round-trips are high-latency; read ahead on demand rather than prefilling the working set.
- **`uri_parser`: S3 object keys are literal bytes.** `%`, `?` and `#` are ordinary key bytes, matching AWS CLI semantics, so `s3://b/a%20b` now opens the key `a%20b`. Previously the key was percent-decoded and split on `?`, which opened the wrong object.

## New capability

- **S3 ListObjectsV2** — a hand-rolled, fail-closed list parser (no XML dependency); `presign_url(extra_canonical_query)` so caller query params participate in the signature (required for S3 to accept a presigned LIST); `authorize_list` on the authorizer interface and both SigV4 implementations; an `X-Amz-*` injection guard so callers cannot smuggle or override signing params; `rest_reactor::list_page`; and `rest_ioctx::list_objects_paged` / `list_objects` / `list_max_matches`. Pagination is guarded against non-conforming backends: truncated-without-token, truncated-but-empty, and non-advancing continuation token all throw rather than loop or silently truncate.
- **Parquet footer probe** — `open_hint` plus `create_io_object(path, hint)`; a suffix-range GET that resolves the object size *and* stashes its trailing bytes in one round-trip; and a stash-hit fast path in `rest_reactor::host_read` so the footer reads that follow are served from memory instead of costing extra round-trips. Falls back to a plain HEAD on any unusable response (200 full body, 416, missing/unsatisfied `Content-Range`).
- **Known-size open** — `create_io_object(path, known_size)`, so a size already learned from a LIST response builds the io_object with zero network: no HEAD, no probe.
- **`kvikio_context`** — a local-file fallback backend built directly on `kvikio::FileHandle` rather than `cudf::io::datasource`, which keeps the io library cudf-free. Registered as a catch-all that `lookup_path` defers behind the explicit uring/rest backends, so `s3://` never resolves to it and a local file still routes to uring first.
- **`kvikio_config`** — optional-per-field tunables: `nthreads`, `task_size`, `gds_threshold`, `bounce_buffer_size`, O_DIRECT reads and overread, per-block-device pools, and compat mode. An unset field means "leave kvikIO's own default alone", so the `KVIKIO_*` environment variables keep working as the outer default and this is an explicit in-process override layered on top. Every field except `compat_mode` maps to a setter on kvikIO's process-global `defaults` singleton and is documented as such — last context constructed wins. `compat_mode` is the exception: it rides the `FileHandle` constructor, so it scopes to files this ioctx opens and mutates nothing global.

## Cleanup

- `metadata_store::get_metadata(cache_key)` overload, for callers that know the path but have not built an io_object yet.
- Dropped `object_store_config::s3_use_async_backend` — no consumers, and it referenced types that do not exist.
- Renamed `io_context_registry::lookup` → `lookup_path`. It is passed a full path, not a bare scheme (the checkers parse the URI / stat the filesystem themselves), and the docs claimed otherwise.

## Tests

New `cucascade_io_tests` target — **732 assertions across 71 cases**, covering `uri_parser`, sigv4, the list parser, both authorizers, static credentials and `kvikio_config`. **The io layer had no tests at all before this.**

Full suite: **6,330 assertions / 343 cases, 0 failures.**

| Binary | Result |
|---|---|
| `cucascade_tests` | 3225 / 89 |
| `cucascade_io_tests` (new) | 732 / 71 |
| `cucascade_cudf_tests` | 2326 / 177 |
| `cucascade_topology_discovery_tests` | 47 / 6 |

Builds clean with `CUCASCADE_WARNINGS_AS_ERRORS=ON`; `pre-commit run -a` passes.

## Not verified

The two S3 benchmarks are gated behind `CUCASCADE_BUILD_S3_BENCHMARK`, which requires aws-sdk-cpp (present only in the `s3-bench` pixi environment), so **they were not compiled**. By inspection they use only the single-argument `open_datasource` / `open_io_object` overloads — unchanged, since the new behavior was added as overloads rather than signature changes — and nothing switches over `io_context_type`, so the new `kvikio` enumerator cannot trip `-Werror=switch`.

## Follow-up

REST perf instrumentation (per-chunk timings, queue wait, TTFB, retry/terminal counters, pool aggregation) is deliberately not part of this PR. The control-plane paths added here leave room for it and it layers on cleanly; the existing S3 benchmarks are its natural consumer.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Authors:
  - Amin Aramoon (https://github.com/aminaramoon)

Approvers:
  - https://github.com/felipeblazing

URL: #172

Author: aminaramoon

CMakeLists.txt

@@ -158,6 +158,11 @@ if(NOT CUCASCADE_TOPOLOGY_ONLY)
     pkg_check_modules(CURL REQUIRED IMPORTED_TARGET libcurl)
     find_package(OpenSSL REQUIRED)
 
+    # kvikIO — backs the local-file fallback ioctx (kvikio_context). Used
+    # directly (not via cudf) so the io library stays cudf-free. Not swappable:
+    # unlike moodycamel/invocable below there is no in-tree stand-in to replace.
+    find_package(kvikio REQUIRED CONFIG)
+
     # cucascade_io_thirdparty carries the swappable moodycamel + invocable
     # (abseil) usage requirements from a single place; the io object library,
     # its installable static/shared variants, and their in-tree consumers
@@ -402,7 +407,7 @@ if(CUCASCADE_BUILD_IO)
   # side by cuCascadeConfig.cmake (same names), mirroring the Numa::Numa
   # approach.
   set(CUCASCADE_IO_LINK_LIBS PkgConfig::LIBURING PkgConfig::CURL
-                             OpenSSL::Crypto)
+                             OpenSSL::Crypto kvikio::kvikio)
 
   target_link_libraries(
     cucascade_io_objects PUBLIC cucascade_objects ${CUCASCADE_IO_LINK_LIBS}

include/cucascade/cudf/datasource.hpp

@@ -159,4 +159,19 @@ class datasource : public cudf::io::datasource {
 [[nodiscard]] std::unique_ptr<datasource> open_datasource(std::shared_ptr<ioctx> io_ctx,
                                                           std::string path);
 
+/// As above, forwarding @p hint to the backend's io_object resolution so it can,
+/// e.g., prefetch a parquet footer in the same round-trip as the size
+/// (@c open_hint::parquet_footer_probe).  Backends that cannot act on the hint
+/// fall back to the plain open.
+[[nodiscard]] std::unique_ptr<datasource> open_datasource(std::shared_ptr<ioctx> io_ctx,
+                                                          std::string path,
+                                                          open_hint hint);
+
+/// As above, with the object's size already known (e.g. from an S3
+/// ListObjectsV2 response), so a backend that can act on it skips its size
+/// discovery entirely (no HEAD for object stores).
+[[nodiscard]] std::unique_ptr<datasource> open_datasource(std::shared_ptr<ioctx> io_ctx,
+                                                          std::string path,
+                                                          std::uint64_t known_size);
+
 }  // namespace cucascade::io

include/cucascade/io/cache/metadata_store.hpp

@@ -20,13 +20,34 @@
 
 #include <cucascade/io/types.hpp>
 
+#include <cstddef>
+#include <functional>
 #include <memory>
 #include <shared_mutex>
 #include <string>
+#include <string_view>
 #include <unordered_map>
 
 namespace cucascade::io::cache {
 
+namespace detail {
+
+/// Transparent hasher so the store can be looked up by @c std::string_view (or
+/// @c const char*) without materialising a @c std::string.  Paired with
+/// @c std::equal_to<> below, this enables C++20 heterogeneous lookup on the
+/// underlying @c unordered_map — without both, a string_view-taking getter
+/// would just construct a temporary key on every call and be strictly worse
+/// than taking @c std::string const&.
+struct string_hash {
+  using is_transparent = void;
+  [[nodiscard]] std::size_t operator()(std::string_view sv) const noexcept
+  {
+    return std::hash<std::string_view>{}(sv);
+  }
+};
+
+}  // namespace detail
+
 /**
  * @brief Thread-safe per-file metadata cache, keyed by an io_object's
  *        raw_file_cache_id().
@@ -58,9 +79,19 @@ class metadata_store {
   /// miss.
   [[nodiscard]] std::shared_ptr<io_object_metadata> get_metadata(io_object const& obj) const;
 
+  /// As above but keyed directly by @c raw_file_cache_id() — for callers that
+  /// know the path but have not built an io_object yet.  Returns nullptr on miss.
+  /// Looked up heterogeneously, so passing a @c string_view or a string literal
+  /// allocates nothing.
+  [[nodiscard]] std::shared_ptr<io_object_metadata> get_metadata(std::string_view cache_key) const;
+
  private:
   mutable std::shared_mutex _mtx;
-  std::unordered_map<std::string, std::shared_ptr<io_object_metadata>> _by_key;
+  std::unordered_map<std::string,
+                     std::shared_ptr<io_object_metadata>,
+                     detail::string_hash,
+                     std::equal_to<>>
+    _by_key;
 };
 
 }  // namespace cucascade::io::cache

include/cucascade/io/config.hpp

@@ -18,6 +18,7 @@
 #pragma once
 
 #include <cucascade/io/cache/config.hpp>
+#include <cucascade/io/kvikio/config.hpp>
 #include <cucascade/io/object_store_config.hpp>
 #include <cucascade/io/rest/config.hpp>
 #include <cucascade/io/uring/config.hpp>
@@ -34,6 +35,7 @@ namespace cucascade::io {
  * Sub-configs:
  *  - @c local   — uring reactor tunables (local-disk IO path).
  *  - @c rest    — REST reactor tunables (S3/object-store IO path).
+ *  - @c kvikio  — kvikIO fallback tunables (local-disk catch-all path).
  *  - @c cache   — prefetching cache tunables.
  *  - @c object_store — object-store credentials and endpoint.
  */
@@ -57,6 +59,12 @@ struct io_config {
   /// retry policy, etc.
   rest::config rest{};
 
+  /// kvikIO fallback configuration — thread-pool size, task/bounce sizing,
+  /// O_DIRECT, compat mode.  All fields default to "unset", leaving kvikIO's
+  /// own env-var-seeded defaults in place.  Note these are process-global once
+  /// applied; see @ref kvikio_config.
+  kvikio_config kvikio{};
+
   /// Prefetching cache configuration — in-flight budget, pool sizing,
   /// dispose-after-use policy.
   cache::config cache{};

include/cucascade/io/datasource_factory.hpp

@@ -38,17 +38,13 @@ namespace cucascade::io {
 // ---------------------------------------------------------------------------
 
 /**
- * @brief Thread-safe registry mapping URI schemes to @c ioctx instances.
+ * @brief Thread-safe registry of @c ioctx backends, resolved by full path.
  *
- * The engine constructs a registry at startup and populates it with one
- * @c ioctx per backend (uring / gds / s3 / rdma_s3). The factory looks
- * up the correct backend by URI scheme at datasource-creation time.
- *
- * Scheme matching is case-insensitive: @c register_ioctx and @c lookup both
- * lowercase the scheme before storing / searching, matching the
- * normalization done by @c cucascade::io::parse (RFC 3986 §3.1). Callers may
- * register / look up with any casing — @c register_ioctx("S3", ...) and
- * @c lookup("s3") refer to the same entry.
+ * The engine constructs a registry at startup and registers one entry per
+ * backend (kvikio / uring / restful), each carrying a path-capability checker.
+ * At datasource-creation time @c lookup_path runs the checkers against a full
+ * path (the checkers parse the URI / stat the filesystem themselves) and picks
+ * the backend, preferring an explicit backend over the kvikio catch-all.
  *
  * All operations are safe under concurrent reads; mutations take an exclusive
  * lock but are expected only at engine bootstrap / shutdown.
@@ -75,17 +71,21 @@ class io_context_registry {
   using factory_type        = std::function<std::shared_ptr<io::ioctx>(const config_type&)>;
 
   /**
-   * @brief Register an ioctx for a scheme. Replaces any prior registration
-   *        for the same scheme.
+   * @brief Register an ioctx backend. Replaces any prior registration for the
+   *        same type.
    *
-   * The scheme is lowercased before storage; subsequent @c lookup calls
-   * with any casing of the same scheme resolve to this entry.
-   * @param type    Opaque identifier for the ioctx type. Used by the engine to
-   *                identify the backend.
+   * @param type    Backend identifier (uring / restful / kvikio).
+   * @param checker Decides whether this backend claims a given path.
+   * @param factory Constructs the backend's ioctx; invoked by @c make_ioctx.
    */
   void register_ioctx(io_context_type type, scheme_checker_type checker, factory_type factory);
 
-  std::optional<io_context_type> lookup(std::string_view scheme) const noexcept;
+  /// Resolve the backend for a full @p path (not a bare scheme — the checkers
+  /// parse the URI / stat the filesystem themselves).  Explicit backends
+  /// (uring / restful) take precedence over the kvikio catch-all, so `s3://`
+  /// never resolves to kvikio and a local file routes to uring before the
+  /// universal fallback.  std::nullopt when nothing matches.
+  std::optional<io_context_type> lookup_path(std::string_view path) const noexcept;
 
   std::shared_ptr<ioctx> make_ioctx(io_context_type type) const noexcept;
 

include/cucascade/io/io_context.hpp

@@ -27,6 +27,7 @@
 #include <rmm/cuda_stream_view.hpp>
 
 #include <cstddef>
+#include <cstdint>
 #include <memory>
 #include <optional>
 #include <span>
@@ -36,7 +37,15 @@
 
 namespace cucascade::io {
 
-enum class io_context_type { uring, restful };
+enum class io_context_type { uring, restful, kvikio };
+
+/// Hint passed to @c open_io_object so a backend can tailor how it resolves an
+/// object's metadata.  @c generic resolves the size however is cheapest for the
+/// scheme (a HEAD for object stores).  @c parquet_footer_probe asks the backend
+/// to resolve the size *and* stash the object's trailing bytes in one
+/// round-trip (a suffix-range GET), so the parquet footer reads that follow are
+/// served locally instead of costing extra round-trips.
+enum class open_hint { generic, parquet_footer_probe };
 
 namespace cache {
 class prefetching_cache;
@@ -102,6 +111,22 @@ class ioctx : public std::enable_shared_from_this<ioctx> {
     return create_io_object(std::move(path));
   }
 
+  /// As above, forwarding @p hint to the backend's io_object resolution so it
+  /// can, e.g., prefetch a parquet footer in the same round-trip as the size.
+  [[nodiscard]] std::shared_ptr<io_object> open_io_object(std::string path, open_hint hint)
+  {
+    return create_io_object(std::move(path), hint);
+  }
+
+  /// As above, with the object's size already known (e.g. from an S3
+  /// ListObjectsV2 response), so a backend that can act on it skips its size
+  /// discovery entirely (no HEAD for object stores).
+  [[nodiscard]] std::shared_ptr<io_object> open_io_object(std::string path,
+                                                          std::uint64_t known_size)
+  {
+    return create_io_object(std::move(path), known_size);
+  }
+
   /// Whether this backend can serve reads for @p path.  Backends should
   /// validate scheme/protocol support and any backend-specific
   /// preconditions (e.g. file existence for local-disk backends).
@@ -263,6 +288,20 @@ class ioctx : public std::enable_shared_from_this<ioctx> {
   /// on unsupported / unreachable paths.
   virtual std::shared_ptr<io_object> create_io_object(std::string path) = 0;
 
+  /// Hinted variant.  The base implementation ignores @p hint and delegates to
+  /// the required @c create_io_object(path); a backend that can act on the hint
+  /// (e.g. rest_ioctx's suffix-range footer probe) overrides this.  Kept a
+  /// distinct virtual — not a defaulted argument on the pure virtual above — so
+  /// the hint dispatches on the dynamic type instead of binding statically.
+  virtual std::shared_ptr<io_object> create_io_object(std::string path, open_hint hint);
+
+  /// Known-size variant.  The base implementation ignores @p known_size and
+  /// delegates to the required @c create_io_object(path); a backend whose size
+  /// discovery would otherwise cost a round-trip overrides this to build the
+  /// io_object without one.  Same distinct-virtual rationale as the hint
+  /// variant above.
+  virtual std::shared_ptr<io_object> create_io_object(std::string path, std::uint64_t known_size);
+
   /// Owned by this ioctx.  Built by @ref initialize_cache, destroyed
   /// by @ref shutdown_cache (or the ioctx destructor as a safety net,
   /// though callers are expected to drive the lifecycle explicitly so

include/cucascade/io/io_request.hpp

@@ -147,9 +147,7 @@ struct device_cpy_request {
 
   // Issue every copy on @p stream (a batch when there is more than one), then
   // record @p event once after the last so a single wait covers them all.
-  cudaError_t copy_async(uint8_t* host_buffer,
-                         [[maybe_unused]] size_t bytes,
-                         cudaEvent_t event = nullptr) noexcept
+  cudaError_t copy_async(uint8_t* host_buffer, size_t bytes, cudaEvent_t event = nullptr) noexcept
   {
     assert(host_buffer != nullptr && "Caller must provide a valid host buffer for the copy.");
     rmm::cuda_set_device_raii device_guard(rmm::cuda_device_id{device_id});
@@ -159,8 +157,24 @@ struct device_cpy_request {
              "Caller must provide a valid device destination buffer for the copy.");
       assert((c.src != nullptr || c.src_off + c.size <= bytes) &&
              "Caller must ensure the copy fits in the host buffer.");
-      uint8_t* src_ptr = c.src != nullptr ? c.src : host_buffer + c.src_off;
-      err              = cudaMemcpyAsync(c.dst, src_ptr, c.size, cudaMemcpyHostToDevice, stream);
+      // Resolve the host source.  The asserts above are compiled out in release,
+      // so validate *before* forming the pointer: for a bounce-staged copy
+      // (c.src == nullptr) the source is host_buffer + c.src_off, but a null
+      // host_buffer or an out-of-range [src_off, src_off + size) would otherwise
+      // produce UB (nullptr + offset) or a wild in-range pointer that the
+      // near-null check below cannot catch.  A null-buffer segment must reach
+      // here as c.src == nullptr, never as a non-null "nullptr + offset" pointer.
+      uint8_t* src_ptr = nullptr;
+      if (c.src != nullptr) {
+        src_ptr = c.src;
+      } else if (host_buffer != nullptr && c.src_off <= bytes && c.size <= bytes - c.src_off) {
+        src_ptr = host_buffer + c.src_off;
+      }
+      if (c.dst == nullptr || src_ptr == nullptr ||
+          reinterpret_cast<std::uintptr_t>(src_ptr) < 4096U) {
+        return cudaErrorInvalidValue;
+      }
+      err = cudaMemcpyAsync(c.dst, src_ptr, c.size, cudaMemcpyHostToDevice, stream);
       if (err != cudaSuccess) { return err; }
     }
     if (event != nullptr) { err = cudaEventRecord(event, stream); }