feat(io): batched footer resolve on rest_ioctx

include/cucascade/exec/admission_control.hpp

@@ -80,6 +80,11 @@ class admission_control {
   /// If @p stop fires during the wait, returns a disengaged slot.
   [[nodiscard]] slot acquire(size_t size, std::stop_token stop = {});
 
+  /// Non-blocking acquire: reserve exactly @p size units if they fit within
+  /// the remaining budget right now, else return a disengaged slot.  The
+  /// oversized-request fallback of acquire() does not apply here.
+  [[nodiscard]] slot try_acquire(size_t size);
+
   /// Block until every outstanding slot has been released (i.e. all issued
   /// tokens are freed and the in-use budget drops back to zero).  Returns
   /// immediately if nothing is currently reserved.  If @p stop fires first,
@@ -93,13 +98,20 @@ class admission_control {
 
   [[nodiscard]] size_t budget() const noexcept { return _budget; }
 
+  /// Bytes currently reserved by live slots.
+  [[nodiscard]] size_t reserved() const;
+
+  /// High-water mark of @ref reserved over this controller's lifetime.
+  [[nodiscard]] size_t peak_reserved() const;
+
  private:
   void release(size_t reserved) noexcept;
 
   const size_t _budget;
   size_t _in_use{0};
+  size_t _peak{0};
   size_t _active_slots{0};
-  std::mutex _mtx;
+  mutable std::mutex _mtx;
   std::condition_variable_any _cv;
 };
 

include/cucascade/io/rest/config.hpp

@@ -113,6 +113,28 @@ struct config {
   /// two axes diverge when a prefix is huge but few keys match, so both exist.
   std::size_t list_max_matches{s3::default_max_list_objects};     // 100'000
   std::size_t list_max_scanned{s3::default_max_scanned_objects};  // 1'000'000
+
+  /// Sentinel for the footer_resolve_* knobs below: derive the value from the
+  /// ioctx shape instead of using an explicit setting.
+  static constexpr std::size_t footer_resolve_auto{static_cast<std::size_t>(-1)};
+
+  /// Concurrency cap for one @c rest_ioctx::resolve_footer_objects batch: at
+  /// most this many probe/HEAD transfers are on the wire at once, and the
+  /// batch's curl multi pools at most this many connections.
+  /// @c footer_resolve_auto derives n_reactors * max_connections at the ioctx;
+  /// 0 disables the API entirely (resolve_footer_objects throws) — the
+  /// rollback switch.
+  std::size_t footer_resolve_max_inflight{footer_resolve_auto};
+
+  /// Aggregate cap (bytes) on live footer payloads across all batches of one
+  /// ioctx.  Each entry reserves @c footer_probe_bytes just before its GET is
+  /// issued; the bytes return when the delivered payload buffer is freed, so
+  /// the cap paces resolve-ahead to how fast the caller drops payloads.
+  /// @c footer_resolve_auto derives 2 * effective-inflight *
+  /// footer_probe_bytes.  An explicit value smaller than
+  /// @c footer_probe_bytes is rejected at resolve time — a sub-window budget
+  /// cannot be honored as a hard cap.
+  std::size_t footer_resolve_stash_budget{footer_resolve_auto};
 };
 
 }  // namespace cucascade::io::rest

include/cucascade/io/rest/rest_ioctx.hpp

@@ -18,15 +18,21 @@
 
 #pragma once
 
+#include <cucascade/exec/admission_control.hpp>
 #include <cucascade/io/rest/rest_reactor.hpp>
 #include <cucascade/io/rest/s3/list_parser.hpp>
 #include <cucascade/io/templated_ioctx.hpp>
 
+#include <condition_variable>
 #include <cstddef>
 #include <cstdint>
+#include <deque>
 #include <functional>
 #include <memory>
+#include <mutex>
 #include <optional>
+#include <span>
+#include <stop_token>
 #include <string>
 #include <string_view>
 #include <vector>
@@ -59,7 +65,8 @@ class rest_ioctx : public templated_ioctx<rest_reactor> {
 
   /// 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.
+  /// Reactor counters are lock-free; the footer-budget gauge takes one short
+  /// mutex.  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
@@ -92,6 +99,35 @@ class rest_ioctx : public templated_ioctx<rest_reactor> {
   /// practice).
   [[nodiscard]] std::size_t list_max_matches() const;
 
+  /// Resolve many objects' footers concurrently.  Per-entry semantics are
+  /// IDENTICAL to @c open_io_object(path, parquet_footer_probe): one verified
+  /// suffix GET; 200/416/unverifiable-206 fall back to a HEAD supplying
+  /// size+tag; the same retry policy per entry, never stalling siblings.  The
+  /// caller's thread drives one curl multi with connection reuse across
+  /// entries, and @p on_result is invoked ON THE CALLER'S THREAD, SERIALLY,
+  /// as each entry lands — completion order, no all-entries barrier.
+  ///
+  /// Every input occurrence is delivered exactly once (duplicates delivered
+  /// per occurrence, disambiguated by index); no callback runs after the
+  /// call returns.  On @p stop, in-flight transfers abort and every
+  /// undelivered entry receives one std::system_error(operation_canceled) —
+  /// including a batch cancelled while queued behind another batch (one
+  /// active batch per ioctx; concurrent calls FIFO-serialize).  If
+  /// @p on_result throws, the remaining entries are cancelled (delivered as
+  /// canceled, their callback throws suppressed) and the first exception is
+  /// rethrown after the sweep.  Throws directly only on submission errors:
+  /// an empty batch, the API disabled via
+  /// @c config::footer_resolve_max_inflight == 0, or an explicit
+  /// @c footer_resolve_stash_budget smaller than @c footer_probe_bytes.  An
+  /// unparsable or non-s3 path is a per-entry error, not a batch error.
+  /// Should the transfer driver itself fail mid-batch (a curl multi error),
+  /// every undelivered entry is delivered as canceled before that failure
+  /// is rethrown — exactly-once holds on every exit path.  This ioctx must
+  /// outlive the call.
+  void resolve_footer_objects(std::span<std::string const> paths,
+                              std::function<void(footer_resolve_result)> const& on_result,
+                              std::stop_token stop = {});
+
  protected:
   /// Backend hook invoked by @c ioctx::open_io_object: parse @p path
   /// (s3://bucket/key), HEAD it for the size, and build a @c rest_io_object.
@@ -114,6 +150,23 @@ class rest_ioctx : public templated_ioctx<rest_reactor> {
   /// footer reads are served locally.  Falls back to a plain HEAD (no stash)
   /// when the response is unusable.
   std::shared_ptr<io_object> create_footer_probe_object(std::string path);
+
+  /// The effective resolve_footer_objects concurrency cap: the configured
+  /// knob, or n_reactors * max_connections under footer_resolve_auto.  0 =
+  /// the API is disabled.
+  [[nodiscard]] std::size_t footer_resolve_inflight_cap() const;
+
+  // Batched-footer-resolve coordination: one active batch per ioctx, later
+  // calls FIFO-parked on the ticket queue (stop-aware — a queued batch whose
+  // token fires is removed without ever becoming active).  _footer_budget is
+  // created in the constructor and never reassigned, so perf_snapshot() may
+  // read it without the mutex.
+  mutable std::mutex _footer_resolve_mutex;
+  std::condition_variable_any _footer_resolve_cv;
+  std::deque<std::uint64_t> _footer_resolve_queue;
+  std::uint64_t _footer_resolve_next_ticket{0};
+  bool _footer_resolve_active{false};
+  std::shared_ptr<exec::admission_control> _footer_budget;
 };
 
 }  // namespace cucascade::io::rest

include/cucascade/io/rest/rest_reactor.hpp

@@ -18,6 +18,7 @@
 
 #pragma once
 
+#include <cucascade/exec/admission_control.hpp>
 #include <cucascade/io/cache/types.hpp>
 #include <cucascade/io/concurrent_queue.hpp>
 #include <cucascade/io/rest/authorizer.hpp>
@@ -32,6 +33,8 @@
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
+#include <exception>
+#include <functional>
 #include <memory>
 #include <optional>
 #include <span>
@@ -115,6 +118,31 @@ struct head_object_result {
   std::string etag;
 };
 
+// ---------------------------------------------------------------------------
+// footer_resolve_result
+// ---------------------------------------------------------------------------
+
+/// One resolved entry of a batched footer resolve
+/// (@c rest_ioctx::resolve_footer_objects).  Exactly one of {object, error}
+/// is set.
+///
+/// @c object is stashless — identity, size and validation tag only.  The
+/// suffix window bytes arrive in @c footer instead, whose buffer is the byte
+/// lease: it is allocated against the ioctx-wide footer budget
+/// (@c config::footer_resolve_stash_budget) and the bytes return to that
+/// budget when the buffer is freed, so the intended lifetime is parse-only.
+/// Reads on @c object inside the window re-GET over the network.  @c footer
+/// is null on the HEAD-fallback path (probe unusable), where @c window_lo
+/// stays 0.
+struct footer_resolve_result {
+  std::size_t index{0};               ///< position in the submitted span
+  std::string path;                   ///< the submitted path, verbatim
+  std::shared_ptr<io_object> object;  ///< stashless: size + validation tag
+  shared_byte_span footer;            ///< suffix window bytes (the lease)
+  std::size_t window_lo{0};           ///< file offset of footer->front()
+  std::exception_ptr error;           ///< per-entry failure, isolated
+};
+
 // ---------------------------------------------------------------------------
 // rest_io_object
 // ---------------------------------------------------------------------------
@@ -219,6 +247,11 @@ struct rest_perf_snapshot {
   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};
+  // Ioctx-level (not summed across reactors): live / high-water bytes reserved
+  // from the footer-resolve stash budget.  0 when the batched footer API has
+  // never run on this ioctx.
+  std::uint64_t footer_stash_reserved_bytes{0};
+  std::uint64_t footer_stash_reserved_peak_bytes{0};
 };
 
 /// How @c prep_host_rx_request attributes the resulting GETs in the perf
@@ -356,6 +389,29 @@ class rest_reactor {
   /// the caller falls back to a HEAD.  @p bucket / @p key identify the object.
   footer_probe fetch_footer_suffix(std::string_view bucket, std::string_view key, std::size_t n);
 
+  /// Batched footer resolve engine: every entry gets the same per-attempt
+  /// semantics as @c fetch_footer_suffix plus the HEAD fallback, but all
+  /// entries share one curl multi driven on the caller's thread, so
+  /// connections are reused across entries (at most @p max_inflight pooled)
+  /// and at most @p max_inflight transfers are on the wire at once.
+  /// @p paths / @p objects / @p indices are parallel: @p indices carries each
+  /// entry's position in the caller's original batch.  Each probe reserves
+  /// @c footer_probe_bytes from @p budget before its GET is issued
+  /// (non-blocking while any transfer is active; a blocking, stop-aware wait
+  /// only when none is) and the delivered payload buffer carries the
+  /// reservation until it is freed.  @p on_result runs on the caller's
+  /// thread, serially, as entries land; see
+  /// @c rest_ioctx::resolve_footer_objects for the delivery contract.
+  /// Assumes non-empty input and max_inflight >= 1; concurrent-batch
+  /// serialization is the ioctx's job, not this method's.
+  void resolve_footer_batch(std::span<std::string const> paths,
+                            std::span<object_ref const> objects,
+                            std::span<std::size_t const> indices,
+                            std::size_t max_inflight,
+                            std::shared_ptr<exec::admission_control> budget,
+                            std::function<void(footer_resolve_result)> const& on_result,
+                            std::stop_token stop);
+
   /// Blocking bucket-level ListObjectsV2 GET for one page: returns the raw XML
   /// body on HTTP 200.  @p canonical_query is the pre-encoded, key-sorted
   /// request query (no auth params — authorization is added via

src/exec/admission_control.cpp

@@ -18,6 +18,8 @@
 
 #include <cucascade/exec/admission_control.hpp>
 
+#include <algorithm>
+
 namespace cucascade::exec {
 
 admission_control::admission_control(size_t budget) noexcept : _budget(budget) {}
@@ -45,10 +47,33 @@ admission_control::slot admission_control::acquire(size_t size, std::stop_token
     reserved = _budget;
   }
   _in_use += reserved;
+  _peak = std::max(_peak, _in_use);
   ++_active_slots;
   return slot{this, reserved};
 }
 
+admission_control::slot admission_control::try_acquire(size_t size)
+{
+  std::lock_guard lk(_mtx);
+  if (_in_use + size > _budget) { return {}; }
+  _in_use += size;
+  _peak = std::max(_peak, _in_use);
+  ++_active_slots;
+  return slot{this, size};
+}
+
+size_t admission_control::reserved() const
+{
+  std::lock_guard lk(_mtx);
+  return _in_use;
+}
+
+size_t admission_control::peak_reserved() const
+{
+  std::lock_guard lk(_mtx);
+  return _peak;
+}
+
 bool admission_control::wait_for_all(std::stop_token stop)
 {
   std::unique_lock lk(_mtx);