diff --git a/include/cucascade/exec/admission_control.hpp b/include/cucascade/exec/admission_control.hpp index 69e6cda..f9f373f 100644 --- a/include/cucascade/exec/admission_control.hpp +++ b/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; }; diff --git a/include/cucascade/io/rest/config.hpp b/include/cucascade/io/rest/config.hpp index 3eba08d..8629770 100644 --- a/include/cucascade/io/rest/config.hpp +++ b/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(-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 diff --git a/include/cucascade/io/rest/rest_ioctx.hpp b/include/cucascade/io/rest/rest_ioctx.hpp index 9701570..3f05bdd 100644 --- a/include/cucascade/io/rest/rest_ioctx.hpp +++ b/include/cucascade/io/rest/rest_ioctx.hpp @@ -18,15 +18,21 @@ #pragma once +#include #include #include #include +#include #include #include +#include #include #include +#include #include +#include +#include #include #include #include @@ -59,7 +65,8 @@ class rest_ioctx : public templated_ioctx { /// 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 { /// 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 paths, + std::function 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 { /// footer reads are served locally. Falls back to a plain HEAD (no stash) /// when the response is unusable. std::shared_ptr 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 _footer_resolve_queue; + std::uint64_t _footer_resolve_next_ticket{0}; + bool _footer_resolve_active{false}; + std::shared_ptr _footer_budget; }; } // namespace cucascade::io::rest diff --git a/include/cucascade/io/rest/rest_reactor.hpp b/include/cucascade/io/rest/rest_reactor.hpp index 5e843fa..c397e44 100644 --- a/include/cucascade/io/rest/rest_reactor.hpp +++ b/include/cucascade/io/rest/rest_reactor.hpp @@ -18,6 +18,7 @@ #pragma once +#include #include #include #include @@ -32,6 +33,8 @@ #include #include #include +#include +#include #include #include #include @@ -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 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 paths, + std::span objects, + std::span indices, + std::size_t max_inflight, + std::shared_ptr budget, + std::function 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 diff --git a/src/exec/admission_control.cpp b/src/exec/admission_control.cpp index da58526..9fe937a 100644 --- a/src/exec/admission_control.cpp +++ b/src/exec/admission_control.cpp @@ -18,6 +18,8 @@ #include +#include + 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); diff --git a/src/io/rest/rest_ioctx.cpp b/src/io/rest/rest_ioctx.cpp index 6fc6fb5..61782b3 100644 --- a/src/io/rest/rest_ioctx.cpp +++ b/src/io/rest/rest_ioctx.cpp @@ -23,9 +23,13 @@ #include #include #include +#include +#include #include #include +#include #include +#include namespace cucascade::io::rest { @@ -34,6 +38,165 @@ rest_ioctx::rest_ioctx(std::size_t n_reactors, std::shared_ptr(ctx, "rest-" + std::to_string(i++)); }) { + // Created once here and never reassigned: payload leases capture it by + // shared_ptr (so they may outlive this ioctx) and perf_snapshot() reads it + // without the coordination mutex. + if (std::size_t const inflight = footer_resolve_inflight_cap(); inflight > 0) { + auto const& cfg = _reactors.front()->get_config(); + std::size_t bytes = cfg.footer_resolve_stash_budget; + if (bytes == config::footer_resolve_auto) { bytes = 2 * inflight * cfg.footer_probe_bytes; } + _footer_budget = std::make_shared(std::max(bytes, 1)); + } +} + +std::size_t rest_ioctx::footer_resolve_inflight_cap() const +{ + if (_reactors.empty()) { return 0; } + auto const& cfg = _reactors.front()->get_config(); + if (cfg.footer_resolve_max_inflight == config::footer_resolve_auto) { + return std::max(1, _reactors.size() * cfg.max_connections); + } + return cfg.footer_resolve_max_inflight; +} + +void rest_ioctx::resolve_footer_objects(std::span paths, + std::function const& on_result, + std::stop_token stop) +{ + if (paths.empty()) { + throw std::invalid_argument("rest_ioctx::resolve_footer_objects: empty batch"); + } + if (_reactors.empty()) { + throw std::runtime_error("rest_ioctx::resolve_footer_objects: no reactors"); + } + std::size_t const max_inflight = footer_resolve_inflight_cap(); + if (max_inflight == 0 || !_footer_budget) { + throw std::invalid_argument( + "rest_ioctx::resolve_footer_objects: disabled (footer_resolve_max_inflight == 0)"); + } + if (_footer_budget->budget() < _reactors.front()->get_config().footer_probe_bytes) { + // A budget below one probe window cannot admit any entry without + // over-committing past the cap, so it cannot be honored as a hard cap. + throw std::invalid_argument( + "rest_ioctx::resolve_footer_objects: footer_resolve_stash_budget smaller than " + "footer_probe_bytes"); + } + + // Parse up front; a bad scheme is a per-entry error (isolation), not a + // batch error. + std::vector valid_paths; + std::vector valid_objects; + std::vector valid_indices; + std::vector> parse_errors; + valid_paths.reserve(paths.size()); + valid_objects.reserve(paths.size()); + valid_indices.reserve(paths.size()); + for (std::size_t i = 0; i < paths.size(); ++i) { + try { + auto parsed = cucascade::io::parse(paths[i]); + if (parsed.scheme != "s3") { + throw std::invalid_argument("rest_ioctx::resolve_footer_objects: unsupported scheme '" + + parsed.scheme + "'"); + } + valid_paths.push_back(paths[i]); + valid_objects.push_back(object_ref{std::move(parsed.host), std::move(parsed.path)}); + valid_indices.push_back(i); + } catch (...) { + // A malformed path joins bad-scheme paths as a per-entry error — + // parsing must never cost the batch its exactly-once delivery. + parse_errors.emplace_back(i, std::current_exception()); + } + } + + std::exception_ptr callback_error; + auto deliver_guarded = [&](footer_resolve_result&& r) { + try { + on_result(std::move(r)); + } catch (...) { + // First exception wins; later throws during a cancel sweep are + // suppressed. + if (!callback_error) { callback_error = std::current_exception(); } + } + }; + auto canceled = [] { + return std::make_exception_ptr( + std::system_error(std::make_error_code(std::errc::operation_canceled), + "rest_ioctx::resolve_footer_objects: canceled")); + }; + + // FIFO admission: one active batch per ioctx; the wait is stop-aware, so a + // queued batch whose token fires is removed without ever becoming active. + { + std::unique_lock lk(_footer_resolve_mutex); + std::uint64_t const ticket = _footer_resolve_next_ticket++; + _footer_resolve_queue.push_back(ticket); + bool const admitted = _footer_resolve_cv.wait(lk, stop, [&] { + return !_footer_resolve_active && !_footer_resolve_queue.empty() && + _footer_resolve_queue.front() == ticket; + }); + if (!admitted) { + std::erase(_footer_resolve_queue, ticket); + lk.unlock(); + _footer_resolve_cv.notify_all(); + for (std::size_t i = 0; i < paths.size(); ++i) { + footer_resolve_result r; + r.index = i; + r.path = paths[i]; + r.error = canceled(); + deliver_guarded(std::move(r)); + } + if (callback_error) { std::rethrow_exception(callback_error); } + return; + } + _footer_resolve_active = true; + _footer_resolve_queue.pop_front(); + } + + struct gate_release { + rest_ioctx* self; + ~gate_release() + { + { + std::lock_guard lk(self->_footer_resolve_mutex); + self->_footer_resolve_active = false; + } + self->_footer_resolve_cv.notify_all(); + } + } release{this}; + + // Deliver parse failures first (serial, on this thread); if a callback + // throws, every not-yet-delivered entry is swept as canceled and the first + // exception is rethrown — same rule as the engine. + std::size_t parse_pos = 0; + for (; parse_pos < parse_errors.size() && !callback_error; ++parse_pos) { + footer_resolve_result r; + r.index = parse_errors[parse_pos].first; + r.path = paths[parse_errors[parse_pos].first]; + r.error = std::move(parse_errors[parse_pos].second); + deliver_guarded(std::move(r)); + } + if (callback_error) { + for (std::size_t p = parse_pos; p < parse_errors.size(); ++p) { + footer_resolve_result r; + r.index = parse_errors[p].first; + r.path = paths[parse_errors[p].first]; + r.error = canceled(); + deliver_guarded(std::move(r)); + } + for (std::size_t v = 0; v < valid_indices.size(); ++v) { + footer_resolve_result r; + r.index = valid_indices[v]; + r.path = valid_paths[v]; + r.error = canceled(); + deliver_guarded(std::move(r)); + } + std::rethrow_exception(callback_error); + } + + if (valid_indices.empty()) { return; } + + _reactors.front()->resolve_footer_batch( + valid_paths, valid_objects, valid_indices, max_inflight, _footer_budget, on_result, stop); } rest_perf_snapshot rest_ioctx::perf_snapshot() const noexcept @@ -61,6 +224,10 @@ rest_perf_snapshot rest_ioctx::perf_snapshot() const noexcept agg.blocking_host_get_wall_ns_max = std::max(agg.blocking_host_get_wall_ns_max, s.blocking_host_get_wall_ns_max); } + if (_footer_budget) { + agg.footer_stash_reserved_bytes = _footer_budget->reserved(); + agg.footer_stash_reserved_peak_bytes = _footer_budget->peak_reserved(); + } return agg; } diff --git a/src/io/rest/rest_reactor.cpp b/src/io/rest/rest_reactor.cpp index f73d896..0aa0b00 100644 --- a/src/io/rest/rest_reactor.cpp +++ b/src/io/rest/rest_reactor.cpp @@ -38,6 +38,7 @@ #include #include #include +#include #include #include #include @@ -1116,6 +1117,466 @@ footer_probe rest_reactor::fetch_footer_suffix(std::string_view bucket, ") for " + obj.bucket + "/" + obj.key); } +// --------------------------------------------------------------------------- +// batched footer resolve +// --------------------------------------------------------------------------- + +namespace { + +/// Backing storage for a footer payload delivered by resolve_footer_batch: +/// the byte buffer plus the entry's budget reservation. Member order is the +/// release contract — `lease` is declared after `budget`, so it is destroyed +/// first and returns its bytes to a still-alive admission_control even when +/// this storage outlives the ioctx that created it. +struct leased_byte_storage { + std::shared_ptr budget; + exec::admission_control::slot lease; + std::vector bytes; + std::span view; + + leased_byte_storage(std::shared_ptr b, + exec::admission_control::slot l, + std::vector data) + : budget(std::move(b)), lease(std::move(l)), bytes(std::move(data)), view(bytes) + { + } + + leased_byte_storage(leased_byte_storage const&) = delete; + leased_byte_storage& operator=(leased_byte_storage const&) = delete; + leased_byte_storage(leased_byte_storage&&) = delete; + leased_byte_storage& operator=(leased_byte_storage&&) = delete; +}; + +shared_byte_span make_leased_byte_span(std::shared_ptr budget, + exec::admission_control::slot lease, + std::vector bytes) +{ + auto owner = + std::make_shared(std::move(budget), std::move(lease), std::move(bytes)); + return shared_byte_span{owner, &owner->view}; +} + +enum class footer_entry_stage : std::uint8_t { pending, transfer, backoff, done }; +enum class footer_entry_kind : std::uint8_t { probe, head }; + +/// Per-entry state of one batched footer resolve. Lives in a fixed-size +/// vector for the whole batch — the curl callbacks hold pointers into it. +struct footer_entry { + std::size_t pos{0}; // position in the batch's parallel spans + footer_entry_stage stage{footer_entry_stage::pending}; + footer_entry_kind kind{footer_entry_kind::probe}; + std::size_t attempt{0}; + suffix_sink sink; + head_capture head; + exec::admission_control::slot lease; + curl_easy_ptr easy; + curl_slist_ptr headers; + std::string range; + std::string last_error; + std::chrono::steady_clock::time_point retry_at{}; + std::chrono::steady_clock::time_point t0{}; +}; + +} // namespace + +void rest_reactor::resolve_footer_batch(std::span paths, + std::span objects, + std::span indices, + std::size_t max_inflight, + std::shared_ptr budget, + std::function const& on_result, + std::stop_token stop) +{ + std::size_t const window = _config.footer_probe_bytes; + + curl_multi_ptr multi{curl_multi_init()}; + if (!multi) { + throw std::runtime_error("rest_reactor::resolve_footer_batch: curl_multi_init failed"); + } + CUCASCADE_CURLM_CHECK(curl_multi_setopt(multi.get(), CURLMOPT_PIPELINING, CURLPIPE_NOTHING)); + CUCASCADE_CURLM_CHECK( + curl_multi_setopt(multi.get(), CURLMOPT_MAXCONNECTS, static_cast(max_inflight))); + CUCASCADE_CURLM_CHECK( + curl_multi_setopt(multi.get(), CURLMOPT_MAX_HOST_CONNECTIONS, static_cast(max_inflight))); + + // curl_multi_wakeup is the one multi function that is safe to call from + // another thread; a wakeup with no poll in flight makes the next poll + // return early, so the stop signal cannot be lost between the check and + // the poll. + std::stop_callback wake{stop, [&multi] { curl_multi_wakeup(multi.get()); }}; + + std::vector entries(paths.size()); + for (std::size_t i = 0; i < entries.size(); ++i) { + entries[i].pos = i; + // Single-probe parity: fetch_footer_suffix skips the suffix GET entirely + // when the window is zero, so a zero-window batch entry starts at the + // HEAD fallback directly (no GET, no lease). + if (window == 0) { entries[i].kind = footer_entry_kind::head; } + } + + // Unwind safety: should anything below throw while transfers are in + // flight, every easy handle still attached to the multi must be detached + // BEFORE `entries` (which owns the handles) is destroyed — cleaning up an + // easy handle still added to a multi is undefined. Normal exits detach in + // process_completions / cancel_remaining, leaving this a no-op. + struct multi_detach { + CURLM* m; + std::vector* entries; + ~multi_detach() + { + for (auto& e : *entries) { + if (e.easy) { curl_multi_remove_handle(m, e.easy.get()); } + } + } + } detach_guard{multi.get(), &entries}; + + std::size_t undelivered = entries.size(); + std::size_t active = 0; + std::size_t next_to_start = 0; + std::exception_ptr callback_error; + + // Backoff bookkeeping: a count plus a deadline min-heap, so the event loop + // never rescans the whole entry vector. Heap records whose entry left the + // backoff stage are skipped lazily on pop. + std::size_t backoff_count = 0; + using retry_record = std::pair; + std::priority_queue, std::greater<>> retry_heap; + + auto deliver = [&](footer_entry& e, footer_resolve_result&& r) { + e.stage = footer_entry_stage::done; + --undelivered; + try { + on_result(std::move(r)); + } catch (...) { + // First exception wins; throws from the cancel sweep's own deliveries + // are suppressed. + if (!callback_error) { callback_error = std::current_exception(); } + } + }; + + auto error_result = [&](footer_entry const& e, std::exception_ptr err) { + footer_resolve_result r; + r.index = indices[e.pos]; + r.path = paths[e.pos]; + r.error = std::move(err); + return r; + }; + + auto fail_entry = [&](footer_entry& e, std::string const& what) { + _perf.terminal_failures_total.fetch_add(1, std::memory_order_relaxed); + // Buffer before lease: the bytes must be gone before the ledger says so. + e.sink = suffix_sink{}; + e.lease = {}; + deliver(e, + error_result(e, + std::make_exception_ptr(std::runtime_error( + "rest_reactor::resolve_footer_batch: " + what + " for " + + objects[e.pos].bucket + "/" + objects[e.pos].key)))); + }; + + auto cancel_remaining = [&] { + for (auto& e : entries) { + if (e.stage == footer_entry_stage::done) { continue; } + if (e.easy) { + curl_multi_remove_handle(multi.get(), e.easy.get()); + e.easy.reset(); + e.headers.reset(); + if (e.stage == footer_entry_stage::transfer) { --active; } + } + e.sink = suffix_sink{}; + e.lease = {}; + deliver(e, + error_result(e, + std::make_exception_ptr( + std::system_error(std::make_error_code(std::errc::operation_canceled), + "rest_reactor::resolve_footer_batch: canceled")))); + } + }; + + auto submit = [&](footer_entry& e) { + bool const is_probe = e.kind == footer_entry_kind::probe; + try { + auto const authd = + _ctx->authorizer()->authorize(objects[e.pos], + is_probe ? request_method::GET : request_method::HEAD, + presign_ttl(_config)); + + e.easy = curl_easy_ptr{curl_easy_init()}; + if (!e.easy) { + throw std::runtime_error("rest_reactor::resolve_footer_batch: curl_easy_init failed"); + } + configure_easy_handle(e.easy.get(), global_curl_context::instance().share_handle()); + apply_request_opts(e.easy.get(), _config); + if (is_probe) { + e.sink = suffix_sink{}; + e.sink.cap = window; + // Reserve up front so the buffer never grows past the budgeted window + // while bytes stream in. + e.sink.data.reserve(window); + e.range = suffix_range_header(window); + e.headers = build_header_list(authd.headers, &e.range); + CUCASCADE_CURL_CHECK( + curl_easy_setopt(e.easy.get(), CURLOPT_WRITEFUNCTION, &suffix_write_cb)); + CUCASCADE_CURL_CHECK(curl_easy_setopt(e.easy.get(), CURLOPT_WRITEDATA, &e.sink)); + CUCASCADE_CURL_CHECK( + curl_easy_setopt(e.easy.get(), CURLOPT_HEADERFUNCTION, &suffix_header_cb)); + CUCASCADE_CURL_CHECK(curl_easy_setopt(e.easy.get(), CURLOPT_HEADERDATA, &e.sink)); + } else { + e.head = head_capture{}; + e.headers = build_header_list(authd.headers, nullptr); + CUCASCADE_CURL_CHECK(curl_easy_setopt(e.easy.get(), CURLOPT_NOBODY, 1L)); + CUCASCADE_CURL_CHECK(curl_easy_setopt(e.easy.get(), CURLOPT_WRITEFUNCTION, &write_discard)); + CUCASCADE_CURL_CHECK( + curl_easy_setopt(e.easy.get(), CURLOPT_HEADERFUNCTION, &head_header_cb)); + CUCASCADE_CURL_CHECK(curl_easy_setopt(e.easy.get(), CURLOPT_HEADERDATA, &e.head)); + } + CUCASCADE_CURL_CHECK(curl_easy_setopt(e.easy.get(), CURLOPT_URL, authd.url.c_str())); + CUCASCADE_CURL_CHECK(curl_easy_setopt(e.easy.get(), CURLOPT_HTTPHEADER, e.headers.get())); + CUCASCADE_CURL_CHECK(curl_easy_setopt(e.easy.get(), CURLOPT_PRIVATE, &e)); + if (_config.perf_instrumentation) { e.t0 = std::chrono::steady_clock::now(); } + CUCASCADE_CURLM_CHECK(curl_multi_add_handle(multi.get(), e.easy.get())); + e.stage = footer_entry_stage::transfer; + ++active; + } catch (...) { + // Per-entry failure: the authorizer may throw on credential errors and + // any curl setup check may throw; the entry gets that exception and + // siblings continue. The handle is not in the multi here — every check + // above precedes curl_multi_add_handle, and a failed add does not add. + e.easy.reset(); + e.headers.reset(); + e.sink = suffix_sink{}; + e.lease = {}; + deliver(e, error_result(e, std::current_exception())); + } + }; + + auto schedule_retry = [&](footer_entry& e, std::string const& retry_after) { + if (e.attempt + 1 < _config.max_retry_attempts) { + _perf.retries_total.fetch_add(1, std::memory_order_relaxed); + CUCASCADE_LOG_WARN( + "rest_reactor::resolve_footer_batch: retrying {}/{} after {} (attempt {}/{})", + objects[e.pos].bucket, + objects[e.pos].key, + e.last_error, + e.attempt + 1, + _config.max_retry_attempts); + e.retry_at = + std::chrono::steady_clock::now() + compute_backoff(e.attempt, retry_after, _config); + e.attempt += 1; + e.stage = footer_entry_stage::backoff; + ++backoff_count; + retry_heap.emplace(e.retry_at, &e); + } else { + fail_entry(e, "exhausted retries (" + e.last_error + ")"); + } + }; + + auto finish_probe = [&](footer_entry& e, CURLcode rc, long status) { + _perf.payload_bytes_read_total.fetch_add(e.sink.total_received, std::memory_order_relaxed); + if (rc != CURLE_OK && rc != CURLE_WRITE_ERROR) { + e.last_error = std::string(curl_easy_strerror(rc)); + if (!is_retriable_curl(rc)) { + fail_entry(e, e.last_error); + return; + } + schedule_retry(e, e.sink.retry_after); + return; + } + if (status == 206) { + auto const total = content_range_total(e.sink.content_range); + auto const start = content_range_start(e.sink.content_range); + if (total && start && *start <= *total && e.sink.data.size() == *total - *start) { + if (_config.perf_instrumentation) { + auto const get_ns = + static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now() - e.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); + } + footer_resolve_result r; + r.index = indices[e.pos]; + r.path = paths[e.pos]; + r.window_lo = *start; + r.object = std::make_shared( + paths[e.pos], objects[e.pos].bucket, objects[e.pos].key, *total, std::move(e.sink.etag)); + r.footer = make_leased_byte_span(budget, std::move(e.lease), std::move(e.sink.data)); + deliver(e, std::move(r)); + return; + } + // Unverifiable 206: like the blocking path, fall back to a HEAD. The + // body bytes and the lease are both returned — a HEAD delivers no + // payload, and a discarded buffer must not outlive its reservation. + e.sink = suffix_sink{}; + e.lease = {}; + e.kind = footer_entry_kind::head; + e.attempt = 0; + submit(e); + return; + } + if (status == 200 || status == 416) { + e.sink = suffix_sink{}; + e.lease = {}; + e.kind = footer_entry_kind::head; + e.attempt = 0; + submit(e); + return; + } + if (is_retriable_status(status)) { + e.last_error = "HTTP " + std::to_string(status); + schedule_retry(e, e.sink.retry_after); + return; + } + fail_entry(e, "HTTP " + std::to_string(status)); + }; + + auto finish_head = [&](footer_entry& e, CURLcode rc, long status, curl_off_t content_length) { + if (rc == CURLE_OK && status == 200) { + if (content_length < 0) { + fail_entry(e, "missing Content-Length"); + return; + } + footer_resolve_result r; + r.index = indices[e.pos]; + r.path = paths[e.pos]; + r.object = std::make_shared(paths[e.pos], + objects[e.pos].bucket, + objects[e.pos].key, + static_cast(content_length), + std::move(e.head.etag)); + deliver(e, std::move(r)); + return; + } + e.last_error = + rc != CURLE_OK ? std::string(curl_easy_strerror(rc)) : ("HTTP " + std::to_string(status)); + bool const retriable = + (rc != CURLE_OK && is_retriable_curl(rc)) || (rc == CURLE_OK && is_retriable_status(status)); + if (!retriable) { + fail_entry(e, e.last_error); + return; + } + schedule_retry(e, e.head.retry_after); + }; + + auto process_completions = [&] { + int msgs_left = 0; + while (CURLMsg* msg = curl_multi_info_read(multi.get(), &msgs_left)) { + // After the first callback throw (or a stop), nothing more may be + // delivered as success — undrained completions stay attached and fall + // to the cancel sweep. + if (callback_error || stop.stop_requested()) { break; } + if (msg->msg != CURLMSG_DONE) { continue; } + CURL* h = msg->easy_handle; + CURLcode const rc = msg->data.result; + void* priv = nullptr; + curl_easy_getinfo(h, CURLINFO_PRIVATE, &priv); + auto& e = *static_cast(priv); + long status = 0; + curl_easy_getinfo(h, CURLINFO_RESPONSE_CODE, &status); + curl_off_t content_length = -1; + if (e.kind == footer_entry_kind::head) { + curl_easy_getinfo(h, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &content_length); + } + CUCASCADE_CURLM_CHECK(curl_multi_remove_handle(multi.get(), h)); + e.easy.reset(); + e.headers.reset(); + --active; + if (e.kind == footer_entry_kind::probe) { + finish_probe(e, rc, status); + } else { + finish_head(e, rc, status, content_length); + } + } + }; + + // Returns false when a blocking budget wait was cut short by @p stop. + auto start_pending = [&] { + while (!callback_error && !stop.stop_requested() && active < max_inflight && + next_to_start < entries.size()) { + auto& e = entries[next_to_start]; + if (window != 0) { + exec::admission_control::slot lease; + if (active > 0 || backoff_count > 0) { + // Never block on budget while a transfer or a due retry could + // still make progress and release bytes. + lease = budget->try_acquire(window); + if (!lease) { return true; } + } else { + lease = budget->acquire(window, stop); + if (!lease) { return false; } + } + e.lease = std::move(lease); + } + ++next_to_start; + submit(e); + } + return true; + }; + + auto resubmit_due = [&] { + auto const now = std::chrono::steady_clock::now(); + while (!callback_error && !stop.stop_requested() && !retry_heap.empty() && + active < max_inflight) { + auto const [due, e] = retry_heap.top(); + if (e->stage != footer_entry_stage::backoff) { + retry_heap.pop(); + continue; + } + if (due > now) { break; } + retry_heap.pop(); + --backoff_count; + submit(*e); + } + }; + + auto poll_timeout_ms = [&] { + long timeout = 100; + while (!retry_heap.empty() && retry_heap.top().second->stage != footer_entry_stage::backoff) { + retry_heap.pop(); + } + if (!retry_heap.empty()) { + auto const dt = std::chrono::duration_cast( + retry_heap.top().first - std::chrono::steady_clock::now()) + .count(); + timeout = std::min(timeout, std::max(1, static_cast(dt))); + } + return timeout; + }; + + try { + while (undelivered > 0) { + if (stop.stop_requested() || callback_error) { + cancel_remaining(); + break; + } + resubmit_due(); + if (!start_pending()) { + cancel_remaining(); + break; + } + if (undelivered == 0 || stop.stop_requested() || callback_error) { continue; } + int running = 0; + CUCASCADE_CURLM_CHECK(curl_multi_perform(multi.get(), &running)); + process_completions(); + if (undelivered == 0 || stop.stop_requested() || callback_error) { continue; } + int numfds = 0; + CUCASCADE_CURLM_CHECK( + curl_multi_poll(multi.get(), nullptr, 0, static_cast(poll_timeout_ms()), &numfds)); + } + } catch (...) { + // Driver failure (a curl multi error, an allocation failure): deliver + // the cancel sweep first so exactly-once holds even here, then surface + // the driver's own exception, not a callback's. + cancel_remaining(); + throw; + } + + if (callback_error) { std::rethrow_exception(callback_error); } +} + // --------------------------------------------------------------------------- // capabilities / factory // --------------------------------------------------------------------------- diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index dd613aa..7cf84db 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -64,6 +64,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_footer_resolve.cpp io/rest/test_rest_perf_snapshot.cpp io/rest/test_rest_validation_tag.cpp io/rest/test_shared_byte_span.cpp diff --git a/test/io/rest/loopback_range_server.hpp b/test/io/rest/loopback_range_server.hpp index 9912e26..1aa752a 100644 --- a/test/io/rest/loopback_range_server.hpp +++ b/test/io/rest/loopback_range_server.hpp @@ -31,9 +31,11 @@ #include #include #include +#include #include #include #include +#include #include #include #include @@ -41,6 +43,7 @@ #include #include #include +#include #include #include @@ -54,6 +57,8 @@ struct range_fault_policy { bool fail_all_heads{false}; int head_fail_status{503}; std::chrono::milliseconds response_delay{0}; + /// Hold GET responses until this many GET requests have arrived; 0 disables the barrier. + std::size_t get_response_barrier{0}; bool ignore_range_with_200{false}; bool fail_range_with_416{false}; bool malformed_content_range{false}; @@ -76,12 +81,30 @@ struct listed_object { std::uint64_t size{0}; }; +struct scripted_response { + /// Zero selects the server's normal successful response for this method. + int status{0}; + std::chrono::milliseconds delay{0}; + std::optional etag; + std::optional retry_after; + bool malformed_content_range{false}; +}; + +struct key_response_script { + std::vector gets; + std::vector heads; +}; + class loopback_range_server { public: explicit loopback_range_server(std::vector object, - range_fault_policy fault = {}, - std::vector listed = {}) - : _object(std::move(object)), _fault(fault), _listed(std::move(listed)) + range_fault_policy fault = {}, + std::vector listed = {}, + std::unordered_map scripts = {}) + : _object(std::move(object)), + _fault(std::move(fault)), + _listed(std::move(listed)), + _scripts(std::move(scripts)) { if (_object.empty()) { throw std::runtime_error("loopback object must be non-empty"); } @@ -115,6 +138,7 @@ class loopback_range_server { ~loopback_range_server() { _stop.store(true, std::memory_order_relaxed); + _get_response_cv.notify_all(); if (_listen_fd >= 0) { // shutdown() wakes the blocked accept(); close() alone does not reliably interrupt it. ::shutdown(_listen_fd, SHUT_RDWR); @@ -131,10 +155,23 @@ class loopback_range_server { 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::uint16_t port() const noexcept { return _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(); } + [[nodiscard]] std::size_t accepted_connection_count() const noexcept + { + return _accepted_connection_count.load(); + } + [[nodiscard]] std::size_t head_count(std::string_view key) const + { + return request_count(_head_counts_by_key, key); + } + [[nodiscard]] std::size_t get_count(std::string_view key) const + { + return request_count(_get_counts_by_key, key); + } private: static std::string errno_message() { return std::strerror(errno); } @@ -166,6 +203,11 @@ class loopback_range_server { send_all(fd, response); } + static void append_connection_header(std::string& response, bool close_connection) + { + response += close_connection ? "\r\nConnection: close" : "\r\nConnection: keep-alive"; + } + void accept_loop() { while (!_stop.load(std::memory_order_relaxed)) { @@ -176,6 +218,7 @@ class loopback_range_server { if (_stop.load(std::memory_order_relaxed)) { return; } continue; } + _accepted_connection_count.fetch_add(1, std::memory_order_relaxed); std::scoped_lock lock{_workers_mutex}; _workers.emplace_back([this, fd] { handle_client(fd); @@ -190,88 +233,118 @@ class loopback_range_server { 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(n)); + std::string pending; + std::string request; + while (!_stop.load(std::memory_order_relaxed) && read_request(fd, pending, request)) { + if (handle_request(fd, request)) { return; } + } + } - bool const is_head = request.rfind("HEAD ", 0) == 0; - bool const is_get = request.rfind("GET ", 0) == 0; + bool handle_request(int fd, std::string const& request) + { + bool const close_connection = requests_connection_close(request); + 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); } + auto const key = request_key(request); if (is_head) { auto const head_idx = _head_count.fetch_add(1, std::memory_order_relaxed); + auto const key_idx = increment_request_count(_head_counts_by_key, key); + auto const scripted = scripted_step(key, false, key_idx); + delay_response(scripted); send_interim_headers(fd, _fault.interim_head_etag, {}, _fault.interim_head_retry_after); - if (_fault.fail_all_heads || head_idx < _fault.fail_first_heads) { - std::string response = - "HTTP/1.1 " + std::to_string(_fault.head_fail_status) + " Error\r\nContent-Length: 0"; - append_etag_header(response, _fault.failed_head_etag); - append_header(response, "Retry-After", _fault.failed_head_retry_after); - response += "\r\nConnection: close\r\n\r\n"; + bool const scripted_failure = scripted && scripted->status >= 400; + bool const policy_failure = + !scripted && (_fault.fail_all_heads || head_idx < _fault.fail_first_heads); + if (scripted_failure || policy_failure) { + auto const status = scripted_failure ? scripted->status : _fault.head_fail_status; + std::string response = "HTTP/1.1 " + std::to_string(status) + " Error\r\nContent-Length: 0"; + append_etag_header(response, response_etag(scripted, _fault.failed_head_etag)); + append_header( + response, "Retry-After", response_retry_after(scripted, _fault.failed_head_retry_after)); + append_connection_header(response, close_connection); + response += "\r\n\r\n"; send_all(fd, response); - return; + return close_connection; } std::string response = "HTTP/1.1 200 OK\r\nContent-Length: " + std::to_string(_object.size()); - append_etag_header(response, _fault.successful_head_etag); - response += "\r\nConnection: close\r\n\r\n"; + append_etag_header(response, response_etag(scripted, _fault.successful_head_etag)); + append_connection_header(response, close_connection); + response += "\r\n\r\n"; send_all(fd, response); - return; + return close_connection; } if (is_list) { + if (_fault.response_delay.count() > 0) { std::this_thread::sleep_for(_fault.response_delay); } _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; + std::string response = + "HTTP/1.1 200 OK\r\nContent-Type: application/xml\r\nContent-Length: " + + std::to_string(body.size()); + append_connection_header(response, close_connection); + response += "\r\n\r\n" + body; + send_all(fd, response); + return close_connection; } 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; + std::string response{"HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0"}; + append_connection_header(response, close_connection); + response += "\r\n\r\n"; + send_all(fd, response); + return close_connection; } auto const get_idx = _get_count.fetch_add(1, std::memory_order_relaxed); + auto const key_idx = increment_request_count(_get_counts_by_key, key); + wait_at_get_response_barrier(); + auto const scripted = scripted_step(key, true, key_idx); + delay_response(scripted); send_interim_headers(fd, _fault.interim_get_etag, _fault.interim_get_content_range, _fault.interim_get_retry_after); - if (_fault.fail_all_gets || get_idx < _fault.fail_first_gets) { - std::string response = - "HTTP/1.1 " + std::to_string(_fault.fail_status) + " Error\r\nContent-Length: 0"; - append_etag_header(response, _fault.failed_get_etag); - append_header(response, "Retry-After", _fault.failed_get_retry_after); - response += "\r\nConnection: close\r\n\r\n"; + bool const scripted_failure = scripted && scripted->status >= 400; + bool const policy_failure = + !scripted && (_fault.fail_all_gets || get_idx < _fault.fail_first_gets); + if (scripted_failure || policy_failure) { + auto const status = scripted_failure ? scripted->status : _fault.fail_status; + std::string response = "HTTP/1.1 " + std::to_string(status) + " Error\r\nContent-Length: 0"; + append_etag_header(response, response_etag(scripted, _fault.failed_get_etag)); + append_header( + response, "Retry-After", response_retry_after(scripted, _fault.failed_get_retry_after)); + append_connection_header(response, close_connection); + response += "\r\n\r\n"; send_all(fd, response); - return; + return close_connection; } if (auto range = parse_range(request)) { if (_fault.fail_range_with_416) { std::string response{"HTTP/1.1 416 Range Not Satisfiable\r\nContent-Length: 0"}; append_etag_header(response, _fault.failed_get_etag); - response += "\r\nConnection: close\r\n\r\n"; + append_connection_header(response, close_connection); + response += "\r\n\r\n"; send_all(fd, response); - return; + return close_connection; } if (_fault.ignore_range_with_200) { std::string response = "HTTP/1.1 200 OK\r\nContent-Length: " + std::to_string(_object.size()); append_etag_header(response, _fault.failed_get_etag); - response += "\r\nConnection: close\r\n\r\n"; + append_connection_header(response, close_connection); + response += "\r\n\r\n"; send_all(fd, response); send_all(fd, _object.data(), _object.size()); - return; + return close_connection; } auto const [start, end] = *range; auto const size = end - start + 1; std::string response = "HTTP/1.1 206 Partial Content\r\nContent-Length: " + std::to_string(size); - if (_fault.malformed_content_range) { + if (_fault.malformed_content_range || (scripted && scripted->malformed_content_range)) { response += "\r\nContent-Range: bytes malformed"; append_etag_header(response, _fault.failed_get_etag); } else { @@ -279,19 +352,50 @@ class loopback_range_server { response += "\r\nContent-Range: bytes " + std::to_string(start) + "-" + std::to_string(end) + "/" + std::to_string(_object.size()); } - append_etag_header(response, _fault.successful_get_etag); + append_etag_header(response, response_etag(scripted, _fault.successful_get_etag)); } - response += "\r\nConnection: close\r\n\r\n"; + append_connection_header(response, close_connection); + response += "\r\n\r\n"; send_all(fd, response); send_all(fd, _object.data() + start, size); - return; + return close_connection; } std::string response = "HTTP/1.1 200 OK\r\nContent-Length: " + std::to_string(_object.size()); - append_etag_header(response, _fault.successful_get_etag); - response += "\r\nConnection: close\r\n\r\n"; + append_etag_header(response, response_etag(scripted, _fault.successful_get_etag)); + append_connection_header(response, close_connection); + response += "\r\n\r\n"; send_all(fd, response); send_all(fd, _object.data(), _object.size()); + return close_connection; + } + + static bool read_request(int fd, std::string& pending, std::string& request) + { + constexpr std::size_t max_header_bytes = 64 * 1024; + while (true) { + auto const end = pending.find("\r\n\r\n"); + if (end != std::string::npos) { + request = pending.substr(0, end + 4); + pending.erase(0, end + 4); + return true; + } + if (pending.size() >= max_header_bytes) { return false; } + + char bytes[8192]; + ssize_t const n = ::recv(fd, bytes, sizeof(bytes), 0); + if (n <= 0) { return false; } + pending.append(bytes, static_cast(n)); + } + } + + static bool requests_connection_close(std::string const& request) + { + std::string lower = request; + std::transform(lower.begin(), lower.end(), lower.begin(), [](unsigned char c) { + return static_cast(std::tolower(c)); + }); + return lower.find("\r\nconnection: close") != std::string::npos; } static std::string request_target(std::string const& request) @@ -303,6 +407,72 @@ class loopback_range_server { return request.substr(first + 1, second - first - 1); } + static std::string request_key(std::string const& request) + { + auto target = request_target(request); + if (auto const query = target.find('?'); query != std::string::npos) { target.resize(query); } + if (!target.empty() && target.front() == '/') { target.erase(0, 1); } + auto const bucket_end = target.find('/'); + return bucket_end == std::string::npos ? target : target.substr(bucket_end + 1); + } + + [[nodiscard]] std::optional scripted_step(std::string const& key, + bool get, + std::size_t index) const + { + auto const script = _scripts.find(key); + if (script == _scripts.end()) { return std::nullopt; } + auto const& steps = get ? script->second.gets : script->second.heads; + if (steps.empty()) { return std::nullopt; } + return steps[std::min(index, steps.size() - 1)]; + } + + void delay_response(std::optional const& scripted) const + { + auto delay = _fault.response_delay; + if (scripted) { delay += scripted->delay; } + if (delay.count() > 0) { std::this_thread::sleep_for(delay); } + } + + void wait_at_get_response_barrier() + { + if (_fault.get_response_barrier == 0) { return; } + _get_response_cv.notify_all(); + std::unique_lock lock{_get_response_mutex}; + _get_response_cv.wait(lock, [&] { + return _stop.load(std::memory_order_relaxed) || + _get_count.load(std::memory_order_relaxed) >= _fault.get_response_barrier; + }); + } + + static std::string response_etag(std::optional const& scripted, + std::string const& fallback) + { + return scripted && scripted->etag ? *scripted->etag : fallback; + } + + static std::string response_retry_after(std::optional const& scripted, + std::string const& fallback) + { + return scripted && scripted->retry_after ? *scripted->retry_after : fallback; + } + + std::size_t increment_request_count(std::unordered_map& counts, + std::string const& key) + { + std::scoped_lock lock{_request_counts_mutex}; + auto& count = counts[key]; + return count++; + } + + [[nodiscard]] std::size_t request_count( + std::unordered_map const& counts, std::string_view key) const + { + std::scoped_lock lock{_request_counts_mutex}; + auto const found = counts.find(std::string{key}); + return found == counts.end() ? 0 : found->second; + } + static std::string xml_escape(std::string_view value) { std::string out; @@ -392,10 +562,17 @@ class loopback_range_server { std::vector _object; range_fault_policy _fault; std::vector _listed; + std::unordered_map _scripts; std::atomic _stop{false}; + std::atomic _accepted_connection_count{0}; std::atomic _head_count{0}; std::atomic _get_count{0}; std::atomic _list_count{0}; + std::mutex _get_response_mutex; + std::condition_variable _get_response_cv; + mutable std::mutex _request_counts_mutex; + std::unordered_map _head_counts_by_key; + std::unordered_map _get_counts_by_key; std::thread _thread; std::mutex _workers_mutex; std::vector _workers; @@ -405,11 +582,25 @@ class list_capable_mock_authorizer final : public io::rest::request_authorizer { public: explicit list_capable_mock_authorizer(std::string endpoint) : _endpoint(std::move(endpoint)) {} + void set_object_exception(std::string key, std::exception_ptr error) + { + std::scoped_lock lock{_exceptions_mutex}; + _object_exceptions.insert_or_assign(std::move(key), std::move(error)); + } + 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); + std::exception_ptr error; + { + std::scoped_lock lock{_exceptions_mutex}; + if (auto const found = _object_exceptions.find(obj.key); found != _object_exceptions.end()) { + error = found->second; + } + } + if (error) { std::rethrow_exception(error); } return {_endpoint + "/" + obj.bucket + "/" + obj.key, {}}; } @@ -428,6 +619,8 @@ class list_capable_mock_authorizer final : public io::rest::request_authorizer { std::string _endpoint; std::atomic _object_calls{0}; std::atomic _list_calls{0}; + std::mutex _exceptions_mutex; + std::unordered_map _object_exceptions; }; } // namespace cucascade::test diff --git a/test/io/rest/test_rest_footer_resolve.cpp b/test/io/rest/test_rest_footer_resolve.cpp new file mode 100644 index 0000000..2033656 --- /dev/null +++ b/test/io/rest/test_rest_footer_resolve.cpp @@ -0,0 +1,1169 @@ +/* + * 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 +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +using cucascade::exec::scoped_dispatcher; +using cucascade::exec::static_thread_pool; +using cucascade::io::open_hint; +using cucascade::io::rest::config; +using cucascade::io::rest::footer_resolve_result; +using cucascade::io::rest::rest_io_object; +using cucascade::io::rest::rest_ioctx; +using cucascade::io::rest::rest_perf_snapshot; +using cucascade::io::rest::rest_reactor; +using cucascade::io::rest::shared_byte_span; +using cucascade::test::key_response_script; +using cucascade::test::list_capable_mock_authorizer; +using cucascade::test::loopback_range_server; +using cucascade::test::range_fault_policy; +using cucascade::test::scripted_response; +using namespace std::chrono_literals; + +constexpr std::size_t object_size{4096}; +constexpr std::size_t probe_size{256}; + +std::vector test_payload() +{ + std::vector bytes(object_size); + for (std::size_t i = 0; i < bytes.size(); ++i) { + bytes[i] = static_cast((i * 73U + 19U) & 0xffU); + } + return bytes; +} + +std::string object_uri(std::string_view key) { return "s3://bucket/" + std::string{key}; } + +config test_config(std::size_t max_inflight = 2, + std::size_t budget = 4 * probe_size, + bool instrumentation = true) +{ + config cfg{}; + cfg.request_timeout_s = 5; + cfg.tls_verify = false; + cfg.max_connections = 2; + cfg.max_retry_attempts = 3; + cfg.max_auth_retry_attempts = 1; + cfg.retry_backoff_base = 10ms; + cfg.retry_jitter = 0ms; + cfg.honor_retry_after = false; + cfg.perf_instrumentation = instrumentation; + cfg.footer_probe_bytes = probe_size; + cfg.footer_resolve_max_inflight = max_inflight; + cfg.footer_resolve_stash_budget = budget; + return cfg; +} + +struct footer_fixture { + std::shared_ptr authorizer; + std::shared_ptr ioctx; +}; + +footer_fixture make_ioctx(loopback_range_server const& server, config cfg, std::size_t reactors = 1) +{ + auto authorizer = std::make_shared(server.endpoint()); + auto context = std::make_shared(cfg, authorizer, nullptr); + auto ioctx = std::make_shared(reactors, std::move(context)); + return {std::move(authorizer), std::move(ioctx)}; +} + +std::vector resolve(rest_ioctx& ioctx, + std::vector const& paths, + std::stop_token stop = {}) +{ + std::vector results; + results.reserve(paths.size()); + ioctx.resolve_footer_objects( + paths, [&](footer_resolve_result result) { results.push_back(std::move(result)); }, stop); + return results; +} + +footer_resolve_result const& result_at(std::vector const& results, + std::size_t index) +{ + auto const found = std::find_if( + results.begin(), results.end(), [index](auto const& result) { return result.index == index; }); + REQUIRE(found != results.end()); + return *found; +} + +void require_success(footer_resolve_result const& result) +{ + CHECK_FALSE(result.error); + REQUIRE(result.object != nullptr); +} + +void require_failure(footer_resolve_result const& result) +{ + REQUIRE(result.error != nullptr); + CHECK(result.object == nullptr); + CHECK_FALSE(result.footer); +} + +bool is_success(footer_resolve_result const& result) +{ + return !result.error && result.object != nullptr; +} + +void require_span_equal(shared_byte_span const& lhs, shared_byte_span const& rhs) +{ + REQUIRE(static_cast(lhs) == static_cast(rhs)); + if (!lhs) { return; } + REQUIRE(lhs->size() == rhs->size()); + CHECK(std::equal(lhs->begin(), lhs->end(), rhs->begin(), rhs->end())); +} + +bool is_operation_canceled(std::exception_ptr const& error) +{ + if (!error) { return false; } + try { + std::rethrow_exception(error); + } catch (std::system_error const& e) { + return e.code() == std::make_error_code(std::errc::operation_canceled); + } catch (...) { + return false; + } +} + +template +bool wait_until(Predicate&& predicate, std::chrono::milliseconds timeout) +{ + auto const deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (predicate()) { return true; } + std::this_thread::sleep_for(1ms); + } + return predicate(); +} + +class async_call { + public: + template + explicit async_call(Function function) + { + auto promise = std::make_shared>(); + _done = promise->get_future(); + _worker = std::thread([promise, function = std::move(function)]() mutable { + try { + function(); + promise->set_value(nullptr); + } catch (...) { + promise->set_value(std::current_exception()); + } + }); + } + + ~async_call() + { + if (_worker.joinable()) { _worker.detach(); } + } + + async_call(async_call const&) = delete; + async_call& operator=(async_call const&) = delete; + + [[nodiscard]] bool ready_within(std::chrono::milliseconds timeout) + { + return _done.wait_for(timeout) == std::future_status::ready; + } + + std::exception_ptr finish() + { + auto error = _done.get(); + _worker.join(); + return error; + } + + private: + std::future _done; + std::thread _worker; +}; + +void require_ready(async_call& call, std::chrono::milliseconds timeout = 3s) +{ + REQUIRE(call.ready_within(timeout)); + auto const error = call.finish(); + if (error) { std::rethrow_exception(error); } +} + +struct payload_gate { + std::mutex mutex; + std::condition_variable cv; + std::vector payloads; + std::size_t resident_bytes{0}; + std::size_t peak_resident_bytes{0}; + + void retain(shared_byte_span payload) + { + std::scoped_lock lock{mutex}; + resident_bytes += payload ? payload->size() : 0; + peak_resident_bytes = std::max(peak_resident_bytes, resident_bytes); + payloads.push_back(std::move(payload)); + cv.notify_one(); + } + + bool wait_for_payload(std::chrono::milliseconds timeout) + { + std::unique_lock lock{mutex}; + return cv.wait_for(lock, timeout, [&] { return !payloads.empty(); }); + } + + void release_one() + { + shared_byte_span payload; + { + std::scoped_lock lock{mutex}; + if (payloads.empty()) { return; } + payload = std::move(payloads.front()); + payloads.erase(payloads.begin()); + resident_bytes -= payload ? payload->size() : 0; + } + payload.reset(); + cv.notify_all(); + } + + void release_all() + { + while (true) { + { + std::scoped_lock lock{mutex}; + if (payloads.empty()) { return; } + } + release_one(); + } + } + + [[nodiscard]] std::size_t resident() + { + std::scoped_lock lock{mutex}; + return resident_bytes; + } + + [[nodiscard]] std::size_t peak() + { + std::scoped_lock lock{mutex}; + return peak_resident_bytes; + } +}; + +rest_perf_snapshot snapshot_after_single_opens(loopback_range_server const& server, + config cfg, + std::vector const& paths) +{ + auto fixture = make_ioctx(server, cfg); + for (auto const& path : paths) { + try { + (void)fixture.ioctx->open_io_object(path, open_hint::parquet_footer_probe); + } catch (...) { + } + } + return fixture.ioctx->perf_snapshot(); +} + +class per_key_authorizer_error : public std::runtime_error { + public: + using std::runtime_error::runtime_error; +}; + +} // namespace + +TEST_CASE("batched footer resolve matches single-probe bytes, size, and validation tag", + "[rest][footer_resolve]") +{ + SECTION("verified 206") + { + range_fault_policy fault{}; + fault.successful_get_etag = "W/\"footer-v1\""; + loopback_range_server server(test_payload(), fault); + auto fixture = make_ioctx(server, test_config()); + auto const path = object_uri("verified.parquet"); + + auto single_base = fixture.ioctx->open_io_object(path, open_hint::parquet_footer_probe); + auto single = std::dynamic_pointer_cast(single_base); + REQUIRE(single != nullptr); + + auto const results = resolve(*fixture.ioctx, {path}); + auto const& batch = result_at(results, 0); + require_success(batch); + auto batch_object = std::dynamic_pointer_cast(batch.object); + REQUIRE(batch_object != nullptr); + + CHECK(batch.path == path); + CHECK(batch.object->size() == single->size()); + CHECK(batch.object->validation_tag() == single->validation_tag()); + CHECK(batch.window_lo == single->stash_window_lo()); + require_span_equal(batch.footer, single->stash()); + CHECK_FALSE(batch_object->stash()); + } + + SECTION("unusable probe falls back to HEAD") + { + range_fault_policy fault{}; + fault.malformed_content_range = true; + fault.failed_get_etag = "\"discarded\""; + fault.successful_head_etag = "\"head-v1\""; + loopback_range_server server(test_payload(), fault); + auto fixture = make_ioctx(server, test_config()); + auto const path = object_uri("fallback.parquet"); + + auto single_base = fixture.ioctx->open_io_object(path, open_hint::parquet_footer_probe); + auto single = std::dynamic_pointer_cast(single_base); + REQUIRE(single != nullptr); + + auto const results = resolve(*fixture.ioctx, {path}); + auto const& batch = result_at(results, 0); + require_success(batch); + auto batch_object = std::dynamic_pointer_cast(batch.object); + REQUIRE(batch_object != nullptr); + + CHECK(batch.path == path); + CHECK(batch.object->size() == single->size()); + CHECK(batch.object->validation_tag() == single->validation_tag()); + CHECK(batch.object->validation_tag() == "\"head-v1\""); + CHECK(batch.window_lo == single->stash_window_lo()); + require_span_equal(batch.footer, single->stash()); + CHECK_FALSE(batch_object->stash()); + CHECK(server.get_count("fallback.parquet") == 2); + CHECK(server.head_count("fallback.parquet") == 2); + } +} + +TEST_CASE("batched footer resolve uses HEAD only when the probe window is zero", + "[rest][footer_resolve]") +{ + constexpr std::size_t path_count = 3; + range_fault_policy fault{}; + fault.successful_head_etag = "\"head-only\""; + loopback_range_server server(test_payload(), fault); + auto cfg = test_config(path_count); + cfg.footer_probe_bytes = 0; + auto fixture = make_ioctx(server, cfg); + std::vector paths; + paths.reserve(path_count); + for (std::size_t i = 0; i < path_count; ++i) { + paths.push_back(object_uri("head-only-" + std::to_string(i) + ".parquet")); + } + + std::vector results; + results.reserve(paths.size()); + std::uint64_t max_reserved = 0; + fixture.ioctx->resolve_footer_objects(paths, [&](footer_resolve_result result) { + max_reserved = + std::max(max_reserved, fixture.ioctx->perf_snapshot().footer_stash_reserved_bytes); + results.push_back(std::move(result)); + }); + + REQUIRE(results.size() == paths.size()); + for (std::size_t i = 0; i < paths.size(); ++i) { + auto const& result = result_at(results, i); + require_success(result); + CHECK(result.object->size() == object_size); + CHECK(result.object->validation_tag() == "\"head-only\""); + CHECK_FALSE(result.footer); + CHECK(result.window_lo == 0); + CHECK(server.get_count("head-only-" + std::to_string(i) + ".parquet") == 0); + CHECK(server.head_count("head-only-" + std::to_string(i) + ".parquet") == 1); + } + CHECK(server.get_count() == 0); + CHECK(server.head_count() == path_count); + CHECK(max_reserved == 0); + CHECK(fixture.ioctx->perf_snapshot().footer_stash_reserved_bytes == 0); +} + +TEST_CASE("batched footer resolve isolates per-object authorization and not-found errors", + "[rest][footer_resolve]") +{ + std::unordered_map scripts; + scripts["missing.parquet"].gets = {scripted_response{.status = 404}}; + scripts["denied.parquet"].gets = {scripted_response{.status = 403}}; + loopback_range_server server(test_payload(), {}, {}, std::move(scripts)); + auto fixture = make_ioctx(server, test_config(3)); + std::vector paths{object_uri("a.parquet"), + object_uri("missing.parquet"), + object_uri("b.parquet"), + object_uri("denied.parquet")}; + + auto const results = resolve(*fixture.ioctx, paths); + + REQUIRE(results.size() == paths.size()); + for (std::size_t index = 0; index < paths.size(); ++index) { + auto const& result = result_at(results, index); + CHECK(result.path == paths[index]); + if (index == 1 || index == 3) { + require_failure(result); + } else { + require_success(result); + } + } + CHECK(server.get_count("a.parquet") == 1); + CHECK(server.get_count("missing.parquet") == 1); + CHECK(server.get_count("b.parquet") == 1); + CHECK(server.get_count("denied.parquet") == 1); +} + +TEST_CASE("batched footer resolve releases malformed probe buffers before HEAD fallback", + "[rest][footer_resolve]") +{ + constexpr std::size_t malformed_count = 6; + std::unordered_map scripts; + std::vector paths; + for (std::size_t i = 0; i < malformed_count; ++i) { + auto key = "malformed-" + std::to_string(i) + ".parquet"; + scripts[key].gets = {scripted_response{.malformed_content_range = true}}; + paths.push_back(object_uri(key)); + } + paths.push_back(object_uri("normal.parquet")); + + loopback_range_server server(test_payload(), {}, {}, std::move(scripts)); + auto fixture = make_ioctx(server, test_config(2, 2 * probe_size)); + auto results = resolve(*fixture.ioctx, paths); + + REQUIRE(results.size() == paths.size()); + for (std::size_t i = 0; i < malformed_count; ++i) { + auto const& result = result_at(results, i); + require_success(result); + CHECK_FALSE(result.footer); + auto const key = "malformed-" + std::to_string(i) + ".parquet"; + CHECK(server.get_count(key) == 1); + CHECK(server.head_count(key) == 1); + } + auto const& normal = result_at(results, malformed_count); + require_success(normal); + REQUIRE(normal.footer); + CHECK(normal.footer->size() == probe_size); + CHECK(server.get_count("normal.parquet") == 1); + CHECK(server.head_count("normal.parquet") == 0); + + for (auto& result : results) { + result.footer.reset(); + } + CHECK(fixture.ioctx->perf_snapshot().footer_stash_reserved_bytes == 0); +} + +TEST_CASE("batched footer resolve isolates a per-key authorizer exception", + "[rest][footer_resolve]") +{ + loopback_range_server server(test_payload()); + auto fixture = make_ioctx(server, test_config(3)); + fixture.authorizer->set_object_exception( + "auth-error.parquet", + std::make_exception_ptr(per_key_authorizer_error{"original authorizer error"})); + std::vector paths{ + object_uri("a.parquet"), object_uri("auth-error.parquet"), object_uri("b.parquet")}; + + auto const results = resolve(*fixture.ioctx, paths); + + REQUIRE(results.size() == paths.size()); + require_success(result_at(results, 0)); + auto const& failed = result_at(results, 1); + require_failure(failed); + try { + std::rethrow_exception(failed.error); + FAIL("expected the original authorizer exception"); + } catch (per_key_authorizer_error const& error) { + CHECK(std::string_view{error.what()} == "original authorizer error"); + } catch (...) { + FAIL("authorizer exception type changed"); + } + require_success(result_at(results, 2)); + CHECK(server.get_count("a.parquet") == 1); + CHECK(server.get_count("auth-error.parquet") == 0); + CHECK(server.get_count("b.parquet") == 1); + CHECK(fixture.authorizer->object_calls() == 3); +} + +TEST_CASE("batched footer resolve isolates malformed paths and rejects an undersized budget", + "[rest][footer_resolve]") +{ + loopback_range_server server(test_payload()); + auto fixture = make_ioctx(server, test_config(2)); + std::vector paths{"not-a-uri", object_uri("valid.parquet")}; + + auto const results = resolve(*fixture.ioctx, paths); + + REQUIRE(results.size() == paths.size()); + auto const& malformed = result_at(results, 0); + require_failure(malformed); + CHECK_THROWS_AS(std::rethrow_exception(malformed.error), std::invalid_argument); + require_success(result_at(results, 1)); + CHECK(server.get_count("valid.parquet") == 1); + + auto undersized = make_ioctx(server, test_config(1, probe_size - 1)); + std::vector one{object_uri("undersized.parquet")}; + CHECK_THROWS(undersized.ioctx->resolve_footer_objects(one, [](footer_resolve_result) {})); + CHECK(server.get_count("undersized.parquet") == 0); +} + +TEST_CASE("batched footer resolve streams fast siblings while another entry delays and retries", + "[rest][footer_resolve]") +{ + std::unordered_map scripts; + scripts["slow.parquet"].gets = {scripted_response{.delay = 300ms}}; + scripts["retry.parquet"].gets = {scripted_response{.status = 503}, scripted_response{}}; + loopback_range_server server(test_payload(), {}, {}, std::move(scripts)); + auto fixture = make_ioctx(server, test_config(3)); + std::vector paths{ + object_uri("slow.parquet"), object_uri("retry.parquet"), object_uri("fast.parquet")}; + std::vector delivery_order; + std::vector results; + + fixture.ioctx->resolve_footer_objects(paths, [&](footer_resolve_result result) { + delivery_order.push_back(result.index); + results.push_back(std::move(result)); + }); + + REQUIRE(results.size() == 3); + CHECK(std::all_of(results.begin(), results.end(), is_success)); + REQUIRE(delivery_order.size() == 3); + auto sorted_indices = delivery_order; + std::sort(sorted_indices.begin(), sorted_indices.end()); + CHECK((sorted_indices == std::vector{0, 1, 2})); + auto const position = [&](std::size_t index) { + return std::distance(delivery_order.begin(), + std::find(delivery_order.begin(), delivery_order.end(), index)); + }; + CHECK(position(2) < position(1)); + CHECK(position(1) < position(0)); + CHECK(delivery_order.back() == 0); + CHECK(server.get_count("retry.parquet") == 2); + CHECK(fixture.authorizer->object_calls() == 4); +} + +TEST_CASE("batched footer resolve delivers duplicate inputs exactly once on the caller thread", + "[rest][footer_resolve]") +{ + loopback_range_server server(test_payload()); + auto fixture = make_ioctx(server, test_config(2)); + auto const path = object_uri("duplicate.parquet"); + std::vector paths{path, path, path}; + auto const caller = std::this_thread::get_id(); + std::vector results; + std::vector callback_threads; + + fixture.ioctx->resolve_footer_objects(paths, [&](footer_resolve_result result) { + callback_threads.push_back(std::this_thread::get_id()); + results.push_back(std::move(result)); + }); + + REQUIRE(results.size() == paths.size()); + std::vector counts(paths.size()); + for (auto const& result : results) { + require_success(result); + REQUIRE(result.index < counts.size()); + ++counts[result.index]; + CHECK(result.path == path); + } + CHECK((counts == std::vector{1, 1, 1})); + CHECK(std::all_of(callback_threads.begin(), callback_threads.end(), [&](auto thread) { + return thread == caller; + })); + CHECK(server.get_count("duplicate.parquet") == paths.size()); + std::vector empty; + CHECK_THROWS(fixture.ioctx->resolve_footer_objects(empty, [](footer_resolve_result) {})); +} + +TEST_CASE("batched footer resolve cancels every undelivered input exactly once", + "[rest][footer_resolve]") +{ + SECTION("before submission") + { + loopback_range_server server(test_payload()); + auto fixture = make_ioctx(server, test_config()); + std::stop_source stop; + stop.request_stop(); + + auto const results = + resolve(*fixture.ioctx, {object_uri("a.parquet"), object_uri("b.parquet")}, stop.get_token()); + + REQUIRE(results.size() == 2); + CHECK(std::all_of(results.begin(), results.end(), [](auto const& result) { + return is_operation_canceled(result.error); + })); + CHECK(result_at(results, 0).index == 0); + CHECK(result_at(results, 1).index == 1); + require_failure(result_at(results, 0)); + require_failure(result_at(results, 1)); + CHECK(server.get_count() == 0); + } + + SECTION("while GETs are in flight") + { + std::unordered_map scripts; + scripts["a.parquet"].gets = {scripted_response{.delay = 500ms}}; + scripts["b.parquet"].gets = {scripted_response{.delay = 500ms}}; + auto server = std::make_shared( + test_payload(), range_fault_policy{}, std::vector{}, scripts); + auto fixture = make_ioctx(*server, test_config(2)); + auto results = std::make_shared>(); + auto mutex = std::make_shared(); + std::stop_source stop; + std::vector paths{object_uri("a.parquet"), object_uri("b.parquet")}; + async_call call( + [server, ioctx = fixture.ioctx, results, mutex, paths, token = stop.get_token()] { + ioctx->resolve_footer_objects( + paths, + [&](footer_resolve_result result) { + std::scoped_lock lock{*mutex}; + results->push_back(std::move(result)); + }, + token); + }); + + REQUIRE(wait_until([&] { return server->get_count() == 2; }, 1s)); + stop.request_stop(); + require_ready(call, 250ms); + + std::scoped_lock lock{*mutex}; + REQUIRE(results->size() == paths.size()); + CHECK(std::all_of(results->begin(), results->end(), [](auto const& result) { + return is_operation_canceled(result.error); + })); + CHECK(result_at(*results, 0).index == 0); + CHECK(result_at(*results, 1).index == 1); + require_failure(result_at(*results, 0)); + require_failure(result_at(*results, 1)); + } + + SECTION("during retry backoff") + { + std::unordered_map scripts; + scripts["retry.parquet"].gets = {scripted_response{.status = 503}}; + auto server = std::make_shared( + test_payload(), range_fault_policy{}, std::vector{}, scripts); + auto cfg = test_config(1); + cfg.retry_backoff_base = 500ms; + auto fixture = make_ioctx(*server, cfg); + auto results = std::make_shared>(); + std::stop_source stop; + auto const path = object_uri("retry.parquet"); + std::vector paths{path}; + async_call call([server, ioctx = fixture.ioctx, results, paths, token = stop.get_token()] { + ioctx->resolve_footer_objects( + paths, [&](footer_resolve_result result) { results->push_back(std::move(result)); }, token); + }); + + REQUIRE(wait_until([&] { return server->get_count("retry.parquet") == 1; }, 1s)); + stop.request_stop(); + require_ready(call, 250ms); + REQUIRE(results->size() == 1); + CHECK(is_operation_canceled(results->front().error)); + require_failure(results->front()); + CHECK(server->get_count("retry.parquet") == 1); + } +} + +namespace { +class first_callback_error : public std::runtime_error { + public: + first_callback_error() : std::runtime_error("first callback failure") {} +}; + +class later_callback_error : public std::runtime_error { + public: + later_callback_error() : std::runtime_error("later callback failure") {} +}; +} // namespace + +TEST_CASE( + "batched footer resolve drains cancellation callbacks and rethrows the first callback error", + "[rest][footer_resolve]") +{ + struct callback_state { + std::mutex mutex; + std::size_t successful{0}; + std::size_t canceled{0}; + std::size_t invalid_index{0}; + std::array deliveries{}; + bool all_canceled_errors{true}; + }; + + auto server = std::make_shared(test_payload()); + auto fixture = make_ioctx(*server, test_config(1)); + std::vector paths{ + object_uri("a.parquet"), object_uri("b.parquet"), object_uri("c.parquet")}; + auto state = std::make_shared(); + bool caught_first = false; + + async_call call([server, ioctx = fixture.ioctx, paths, state] { + ioctx->resolve_footer_objects(paths, [state](footer_resolve_result result) { + { + std::scoped_lock lock{state->mutex}; + if (result.index < state->deliveries.size()) { + ++state->deliveries[result.index]; + } else { + ++state->invalid_index; + } + } + if (result.error) { + { + std::scoped_lock lock{state->mutex}; + state->all_canceled_errors &= is_operation_canceled(result.error); + ++state->canceled; + } + throw later_callback_error{}; + } + { + std::scoped_lock lock{state->mutex}; + ++state->successful; + } + throw first_callback_error{}; + }); + }); + REQUIRE(call.ready_within(3s)); + auto const error = call.finish(); + try { + if (error) { std::rethrow_exception(error); } + } catch (first_callback_error const&) { + caught_first = true; + } + + CHECK(caught_first); + std::size_t callbacks_at_return; + { + std::scoped_lock lock{state->mutex}; + CHECK(state->successful == 1); + CHECK(state->canceled == paths.size() - 1); + CHECK(state->invalid_index == 0); + CHECK((state->deliveries == std::array{1, 1, 1})); + CHECK(state->all_canceled_errors); + callbacks_at_return = state->successful + state->canceled; + } + std::this_thread::sleep_for(25ms); + std::scoped_lock lock{state->mutex}; + CHECK(state->successful + state->canceled == callbacks_at_return); +} + +TEST_CASE("batched footer resolve cancels concurrent completions after the first callback error", + "[rest][footer_resolve]") +{ + constexpr std::size_t max_inflight = 3; + constexpr std::size_t path_count = 6; + range_fault_policy fault; + fault.get_response_barrier = max_inflight; + loopback_range_server server(test_payload(), fault); + auto fixture = make_ioctx(server, test_config(max_inflight)); + std::vector paths; + paths.reserve(path_count); + for (std::size_t i = 0; i < path_count; ++i) { + paths.push_back(object_uri("callback-" + std::to_string(i) + ".parquet")); + } + + std::array deliveries{}; + std::size_t successes = 0; + std::size_t canceled = 0; + bool all_canceled_errors{true}; + bool caught_first{false}; + try { + fixture.ioctx->resolve_footer_objects(paths, [&](footer_resolve_result result) { + REQUIRE(result.index < deliveries.size()); + ++deliveries[result.index]; + if (result.error) { + all_canceled_errors &= is_operation_canceled(result.error); + ++canceled; + return; + } + ++successes; + throw first_callback_error{}; + }); + } catch (first_callback_error const&) { + caught_first = true; + } + + CHECK(caught_first); + CHECK(successes == 1); + CHECK(canceled == path_count - 1); + CHECK(all_canceled_errors); + CHECK(std::all_of(deliveries.begin(), deliveries.end(), [](auto count) { return count == 1; })); + CHECK(server.get_count() == max_inflight); + for (std::size_t i = 0; i < path_count; ++i) { + auto const key = "callback-" + std::to_string(i) + ".parquet"; + CHECK(server.get_count(key) == (i < max_inflight ? 1 : 0)); + } + CHECK(server.head_count() == 0); + CHECK(fixture.authorizer->object_calls() == max_inflight); + + auto const callbacks_at_return = successes + canceled; + std::this_thread::sleep_for(25ms); + CHECK(successes + canceled == callbacks_at_return); +} + +TEST_CASE("batched footer resolve rejects an explicit zero payload budget", + "[rest][footer_resolve]") +{ + loopback_range_server server(test_payload()); + auto fixture = make_ioctx(server, test_config(2, 0)); + std::vector paths{object_uri("zero-budget.parquet")}; + + CHECK_THROWS_AS(fixture.ioctx->resolve_footer_objects(paths, [](footer_resolve_result) {}), + std::invalid_argument); + CHECK(server.get_count() == 0); + CHECK(server.head_count() == 0); + CHECK(fixture.authorizer->object_calls() == 0); +} + +TEST_CASE("batched footer resolve bounds retained payloads across concurrent batches", + "[rest][footer_resolve]") +{ + constexpr std::size_t budget = probe_size; + auto server = std::make_shared(test_payload()); + auto fixture = make_ioctx(*server, test_config(4, budget)); + auto retained = std::make_shared(); + auto errors = std::make_shared>(0); + auto first_delivered = std::make_shared>(0); + auto second_delivered = std::make_shared>(0); + std::vector first_paths(4, object_uri("first.parquet")); + std::vector second_paths(4, object_uri("second.parquet")); + + auto first_callback = [retained, errors, first_delivered](footer_resolve_result result) { + if (!is_success(result)) { + errors->fetch_add(1); + return; + } + first_delivered->fetch_add(1); + retained->retain(std::move(result.footer)); + }; + auto second_callback = [retained, errors, second_delivered](footer_resolve_result result) { + if (!is_success(result)) { + errors->fetch_add(1); + return; + } + second_delivered->fetch_add(1); + retained->retain(std::move(result.footer)); + }; + async_call first([server, ioctx = fixture.ioctx, first_paths, first_callback] { + ioctx->resolve_footer_objects(first_paths, first_callback); + }); + REQUIRE(wait_until([&] { return server->get_count("first.parquet") == 1; }, 1s)); + async_call second([server, ioctx = fixture.ioctx, second_paths, second_callback] { + ioctx->resolve_footer_objects(second_paths, second_callback); + }); + + bool stayed_within_budget = true; + while (!first.ready_within(0ms)) { + if (!retained->wait_for_payload(1s)) { + throw std::runtime_error("timed out waiting for first-batch footer payload"); + } + stayed_within_budget &= retained->resident() <= budget; + if (first_delivered->load() == first_paths.size()) { break; } + retained->release_one(); + } + require_ready(first); + + REQUIRE(retained->wait_for_payload(1s)); + CHECK(retained->resident() == budget); + CHECK(server->get_count("second.parquet") == 0); + auto const held_snapshot = fixture.ioctx->perf_snapshot(); + CHECK(held_snapshot.footer_stash_reserved_bytes == budget); + CHECK(held_snapshot.footer_stash_reserved_peak_bytes == budget); + retained->release_one(); + REQUIRE(wait_until([&] { return server->get_count("second.parquet") == 1; }, 1s)); + + while (!second.ready_within(0ms)) { + if (!retained->wait_for_payload(1s)) { + throw std::runtime_error("timed out waiting for second-batch footer payload"); + } + stayed_within_budget &= retained->resident() <= budget; + if (second_delivered->load() == second_paths.size()) { break; } + retained->release_one(); + } + require_ready(second); + retained->release_all(); + + auto const drained_snapshot = fixture.ioctx->perf_snapshot(); + CHECK(errors->load() == 0); + CHECK(stayed_within_budget); + CHECK(retained->resident() == 0); + CHECK(retained->peak() <= budget); + CHECK(drained_snapshot.footer_stash_reserved_bytes == 0); + CHECK(drained_snapshot.footer_stash_reserved_peak_bytes == budget); + CHECK(server->get_count() == first_paths.size() + second_paths.size()); + + auto disabled = make_ioctx(*server, test_config(0, budget)); + std::vector one{object_uri("disabled.parquet")}; + CHECK_THROWS(disabled.ioctx->resolve_footer_objects(one, [](footer_resolve_result) {})); +} + +TEST_CASE("batched footer resolve makes progress with a one-worker consumer", + "[rest][footer_resolve]") +{ + auto server = std::make_shared(test_payload()); + auto fixture = make_ioctx(*server, test_config(2, 2 * probe_size)); + auto pool = std::make_shared(1, "footer-one-worker"); + auto dispatcher = std::make_shared(*pool, 1); + auto parsed = std::make_shared>(0); + std::vector paths(12, object_uri("one-worker.parquet")); + + async_call call([server, ioctx = fixture.ioctx, pool, dispatcher, parsed, paths] { + ioctx->resolve_footer_objects(paths, [dispatcher, parsed](footer_resolve_result result) { + dispatcher->enqueue([result = std::move(result), parsed] { + if (!result.error && result.footer) { parsed->fetch_add(1); } + }); + }); + dispatcher->wait_for_all(); + }); + + require_ready(call); + CHECK(parsed->load() == paths.size()); +} + +TEST_CASE("batched footer resolve completes saturated producers and preserves FIFO fairness", + "[rest][footer_resolve]") +{ + auto server = std::make_shared(test_payload()); + auto fixture = make_ioctx(*server, test_config(2, 4 * probe_size)); + auto completed = std::make_shared>(0); + constexpr std::size_t producers = 4; + auto pool = std::make_shared(producers, "footer-saturation"); + auto dispatcher = std::make_shared(*pool, producers); + std::vector> calls; + + for (std::size_t producer = 0; producer < producers; ++producer) { + std::vector paths(4, object_uri("batch-" + std::to_string(producer) + ".parquet")); + calls.push_back(std::make_unique( + [server, ioctx = fixture.ioctx, pool, dispatcher, completed, paths = std::move(paths)] { + ioctx->resolve_footer_objects(paths, [dispatcher, completed](footer_resolve_result result) { + dispatcher->enqueue([result = std::move(result), completed] { + if (!result.error && result.footer) { completed->fetch_add(1); } + }); + }); + })); + } + + for (auto& call : calls) { + require_ready(*call); + } + dispatcher->wait_for_all(); + CHECK(completed->load() == producers * 4); + + std::unordered_map scripts; + scripts["active.parquet"].gets = {scripted_response{.delay = 100ms}}; + auto fifo_server = std::make_shared( + test_payload(), range_fault_policy{}, std::vector{}, scripts); + auto fifo_fixture = make_ioctx(*fifo_server, test_config(1)); + auto order = std::make_shared>(); + auto order_mutex = std::make_shared(); + std::vector active{object_uri("active.parquet")}; + std::vector queued{object_uri("queued.parquet")}; + async_call first([fifo_server, ioctx = fifo_fixture.ioctx, order, order_mutex, active] { + ioctx->resolve_footer_objects(active, [order, order_mutex](footer_resolve_result) { + std::scoped_lock lock{*order_mutex}; + order->push_back('A'); + }); + }); + REQUIRE(wait_until([&] { return fifo_server->get_count("active.parquet") == 1; }, 1s)); + async_call second([fifo_server, ioctx = fifo_fixture.ioctx, order, order_mutex, queued] { + ioctx->resolve_footer_objects(queued, [order, order_mutex](footer_resolve_result) { + std::scoped_lock lock{*order_mutex}; + order->push_back('B'); + }); + }); + + require_ready(first); + require_ready(second); + CHECK((*order == std::vector{'A', 'B'})); +} + +TEST_CASE("batched footer resolve stop unblocks a zero-active-transfer budget wait", + "[rest][footer_resolve]") +{ + auto server = std::make_shared(test_payload()); + auto fixture = make_ioctx(*server, test_config(1, probe_size)); + auto held = std::make_shared>(); + auto results = std::make_shared>(); + auto mutex = std::make_shared(); + std::stop_source stop; + std::vector paths{object_uri("first.parquet"), object_uri("blocked.parquet")}; + async_call call( + [server, ioctx = fixture.ioctx, held, results, mutex, paths, token = stop.get_token()] { + ioctx->resolve_footer_objects( + paths, + [&](footer_resolve_result result) { + std::scoped_lock lock{*mutex}; + if (result.footer) { held->push_back(result.footer); } + results->push_back(std::move(result)); + }, + token); + }); + + REQUIRE(wait_until( + [&] { + std::scoped_lock lock{*mutex}; + return held->size() == 1 && server->get_count("blocked.parquet") == 0; + }, + 1s)); + stop.request_stop(); + require_ready(call, 250ms); + + std::scoped_lock lock{*mutex}; + REQUIRE(results->size() == 2); + require_success(result_at(*results, 0)); + CHECK(is_operation_canceled(result_at(*results, 1).error)); + held->clear(); + results->clear(); +} + +TEST_CASE("batched footer payload lease can outlive its rest ioctx", "[rest][footer_resolve]") +{ + loopback_range_server server(test_payload()); + shared_byte_span held; + auto const expected_last_byte = test_payload().back(); + { + auto fixture = make_ioctx(server, test_config()); + auto results = resolve(*fixture.ioctx, {object_uri("held.parquet")}); + require_success(results.front()); + held = std::move(results.front().footer); + fixture.ioctx.reset(); + } + + REQUIRE(held != nullptr); + REQUIRE_FALSE(held->empty()); + CHECK(held->back() == expected_last_byte); + held.reset(); + SUCCEED(); +} + +TEST_CASE("batched footer resolve removes a canceled batch from the FIFO queue", + "[rest][footer_resolve]") +{ + std::unordered_map scripts; + scripts["active.parquet"].gets = {scripted_response{.delay = 300ms}}; + auto server = std::make_shared( + test_payload(), range_fault_policy{}, std::vector{}, scripts); + auto fixture = make_ioctx(*server, test_config(1)); + std::vector active{object_uri("active.parquet")}; + std::vector queued{object_uri("queued-a.parquet"), object_uri("queued-b.parquet")}; + auto queued_results = std::make_shared>(); + auto queued_started = std::make_shared>(false); + std::stop_source queued_stop; + + async_call first([server, ioctx = fixture.ioctx, active] { + ioctx->resolve_footer_objects(active, [](footer_resolve_result) {}); + }); + REQUIRE(wait_until([&] { return server->get_count("active.parquet") == 1; }, 1s)); + async_call second([server, + ioctx = fixture.ioctx, + queued, + queued_results, + queued_started, + token = queued_stop.get_token()] { + queued_started->store(true); + ioctx->resolve_footer_objects( + queued, + [&](footer_resolve_result result) { queued_results->push_back(std::move(result)); }, + token); + }); + + REQUIRE(wait_until([&] { return queued_started->load(); }, 1s)); + CHECK_FALSE(second.ready_within(20ms)); + queued_stop.request_stop(); + require_ready(second, 1s); + CHECK_FALSE(first.ready_within(0ms)); + REQUIRE(queued_results->size() == queued.size()); + CHECK(std::all_of(queued_results->begin(), queued_results->end(), [](auto const& result) { + return is_operation_canceled(result.error); + })); + CHECK(server->get_count("queued-a.parquet") == 0); + CHECK(server->get_count("queued-b.parquet") == 0); + require_failure(result_at(*queued_results, 0)); + require_failure(result_at(*queued_results, 1)); + require_ready(first); +} + +TEST_CASE("batched footer resolve reuses at most the configured in-flight connections", + "[rest][footer_resolve]") +{ + constexpr std::size_t inflight = 2; + loopback_range_server server(test_payload()); + std::vector paths; + for (std::size_t i = 0; i < 12; ++i) { + paths.push_back(object_uri("reuse-" + std::to_string(i) + ".parquet")); + } + auto fixture = make_ioctx(server, test_config(inflight, paths.size() * probe_size)); + + auto const results = resolve(*fixture.ioctx, paths); + + REQUIRE(results.size() == paths.size()); + CHECK( + std::all_of(results.begin(), results.end(), [](auto const& result) { return !result.error; })); + for (std::size_t i = 0; i < paths.size(); ++i) { + CHECK(server.get_count("reuse-" + std::to_string(i) + ".parquet") == 1); + } + CHECK(server.get_count() == paths.size()); + CHECK(server.accepted_connection_count() <= inflight); +} + +TEST_CASE("batched footer resolve folds equivalent request outcomes into the ioctx snapshot", + "[rest][footer_resolve]") +{ + auto scripts = [] { + std::unordered_map value; + value["retry.parquet"].gets = {scripted_response{.status = 503}, scripted_response{}}; + value["missing.parquet"].gets = {scripted_response{.status = 404}}; + return value; + }; + std::vector paths{ + object_uri("ok.parquet"), object_uri("retry.parquet"), object_uri("missing.parquet")}; + loopback_range_server single_server(test_payload(), {}, {}, scripts()); + auto const single = snapshot_after_single_opens(single_server, test_config(2), paths); + + loopback_range_server batch_server(test_payload(), {}, {}, scripts()); + auto batch_fixture = make_ioctx(batch_server, test_config(2)); + auto const results = resolve(*batch_fixture.ioctx, paths); + auto const batch = batch_fixture.ioctx->perf_snapshot(); + + REQUIRE(results.size() == paths.size()); + require_success(result_at(results, 0)); + require_success(result_at(results, 1)); + require_failure(result_at(results, 2)); + CHECK(batch.chunk_get_count == single.chunk_get_count); + CHECK(batch.blocking_host_get_count == single.blocking_host_get_count); + CHECK(batch.payload_bytes_read_total == single.payload_bytes_read_total); + CHECK(batch.retries_total == single.retries_total); + CHECK(batch.terminal_failures_total == single.terminal_failures_total); +} diff --git a/test/io/rest/test_rest_validation_tag.cpp b/test/io/rest/test_rest_validation_tag.cpp index 8cbad6b..4a3547f 100644 --- a/test/io/rest/test_rest_validation_tag.cpp +++ b/test/io/rest/test_rest_validation_tag.cpp @@ -22,14 +22,21 @@ #include #include +#include #include +#include +#include +#include #include #include #include #include +#include +#include #include #include +#include #include #include @@ -40,8 +47,10 @@ using cucascade::io::rest::config; using cucascade::io::rest::mock_authorizer; using cucascade::io::rest::rest_ioctx; using cucascade::io::rest::rest_reactor; +using cucascade::test::key_response_script; using cucascade::test::loopback_range_server; using cucascade::test::range_fault_policy; +using cucascade::test::scripted_response; using namespace std::chrono_literals; constexpr std::size_t object_size{4096}; @@ -89,8 +98,184 @@ std::shared_ptr make_ioctx(loopback_range_server const& server, conf return std::make_shared(1, std::move(context)); } +struct raw_http_response { + int status{0}; + std::string headers; + std::vector body; + + [[nodiscard]] std::optional header(std::string_view name) const + { + auto const prefix = "\r\n" + std::string{name} + ": "; + auto const begin = headers.find(prefix); + if (begin == std::string::npos) { return std::nullopt; } + auto const value_begin = begin + prefix.size(); + auto const end = headers.find("\r\n", value_begin); + return headers.substr(value_begin, end - value_begin); + } +}; + +class raw_http_connection { + public: + explicit raw_http_connection(loopback_range_server const& server) + { + _fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (_fd < 0) { throw std::runtime_error("socket failed"); } + + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_port = htons(server.port()); + if (::inet_pton(AF_INET, "127.0.0.1", &address.sin_addr) != 1 || + ::connect(_fd, reinterpret_cast(&address), sizeof(address)) != 0) { + ::close(_fd); + _fd = -1; + throw std::runtime_error("connect failed"); + } + } + + ~raw_http_connection() + { + if (_fd >= 0) { ::close(_fd); } + } + + raw_http_connection(raw_http_connection const&) = delete; + raw_http_connection& operator=(raw_http_connection const&) = delete; + + raw_http_response request(std::string_view method, + std::string_view target, + bool close_connection = false) + { + std::string request{method}; + request += " "; + request += target; + request += " HTTP/1.1\r\nHost: 127.0.0.1\r\n"; + if (method == "GET") { request += "Range: bytes=-16\r\n"; } + request += close_connection ? "Connection: close\r\n\r\n" : "Connection: keep-alive\r\n\r\n"; + send_all(request); + return read_response(method != "HEAD"); + } + + [[nodiscard]] bool peer_closed() + { + timeval timeout{}; + timeout.tv_sec = 1; + (void)::setsockopt(_fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); + char byte{}; + return ::recv(_fd, &byte, 1, 0) == 0; + } + + private: + void send_all(std::string_view bytes) + { + std::size_t sent = 0; + while (sent < bytes.size()) { + auto const n = ::send(_fd, bytes.data() + sent, bytes.size() - sent, MSG_NOSIGNAL); + if (n <= 0) { throw std::runtime_error("send failed"); } + sent += static_cast(n); + } + } + + void read_at_least(std::size_t bytes) + { + while (_pending.size() < bytes) { + std::array buffer{}; + auto const n = ::recv(_fd, buffer.data(), buffer.size(), 0); + if (n <= 0) { throw std::runtime_error("unexpected end of HTTP response"); } + _pending.append(buffer.data(), static_cast(n)); + } + } + + raw_http_response read_response(bool has_body) + { + auto header_end = _pending.find("\r\n\r\n"); + while (header_end == std::string::npos) { + read_at_least(_pending.size() + 1); + header_end = _pending.find("\r\n\r\n"); + } + + raw_http_response response; + response.headers = _pending.substr(0, header_end + 4); + _pending.erase(0, header_end + 4); + response.status = std::stoi(response.headers.substr(response.headers.find(' ') + 1, 3)); + + std::size_t body_size = 0; + if (has_body) { + auto const content_length = response.header("Content-Length"); + REQUIRE(content_length.has_value()); + body_size = static_cast(std::stoull(*content_length)); + read_at_least(body_size); + response.body.assign(_pending.begin(), _pending.begin() + body_size); + _pending.erase(0, body_size); + } + return response; + } + + int _fd{-1}; + std::string _pending; +}; + } // namespace +TEST_CASE("loopback range server keeps connections alive and counts requests per key", + "[rest][loopback_server]") +{ + loopback_range_server server(test_payload()); + raw_http_connection connection(server); + + auto const first = connection.request("GET", "/bucket/first.parquet"); + auto const second = connection.request("HEAD", "/bucket/second.parquet", true); + + CHECK(first.status == 206); + CHECK(first.body.size() == 16); + CHECK(second.status == 200); + CHECK(server.accepted_connection_count() == 1); + CHECK(server.get_count() == 1); + CHECK(server.head_count() == 1); + CHECK(server.get_count("first.parquet") == 1); + CHECK(server.get_count("second.parquet") == 0); + CHECK(server.head_count("first.parquet") == 0); + CHECK(server.head_count("second.parquet") == 1); + CHECK(connection.peer_closed()); +} + +TEST_CASE("loopback range server scripts responses independently per key", + "[rest][loopback_server]") +{ + std::unordered_map scripts; + scripts["retry.parquet"].gets = {scripted_response{.status = 503, .etag = "\"retry-v1\""}, + scripted_response{.etag = "\"retry-v2\""}}; + scripts["missing.parquet"].gets = {scripted_response{.status = 404}}; + scripts["denied.parquet"].heads = {scripted_response{.status = 403}}; + scripts["tagged.parquet"].gets = {scripted_response{.etag = "\"tag-v1\""}, + scripted_response{.etag = "\"tag-v2\""}}; + scripts["slow.parquet"].gets = {scripted_response{.delay = 20ms}}; + + loopback_range_server server(test_payload(), {}, {}, std::move(scripts)); + raw_http_connection connection(server); + + CHECK(connection.request("GET", "/bucket/retry.parquet").status == 503); + auto const retry = connection.request("GET", "/bucket/retry.parquet"); + CHECK(retry.status == 206); + CHECK(retry.header("ETag") == "\"retry-v2\""); + CHECK(connection.request("GET", "/bucket/missing.parquet").status == 404); + CHECK(connection.request("HEAD", "/bucket/denied.parquet").status == 403); + + auto const tag_v1 = connection.request("GET", "/bucket/tagged.parquet"); + auto const tag_v2 = connection.request("GET", "/bucket/tagged.parquet"); + CHECK(tag_v1.header("ETag") == "\"tag-v1\""); + CHECK(tag_v2.header("ETag") == "\"tag-v2\""); + + auto const start = std::chrono::steady_clock::now(); + CHECK(connection.request("GET", "/bucket/slow.parquet", true).status == 206); + CHECK(std::chrono::steady_clock::now() - start >= 15ms); + + CHECK(server.get_count("retry.parquet") == 2); + CHECK(server.get_count("missing.parquet") == 1); + CHECK(server.head_count("denied.parquet") == 1); + CHECK(server.get_count("tagged.parquet") == 2); + CHECK(server.get_count("slow.parquet") == 1); + CHECK(server.accepted_connection_count() == 1); +} + TEST_CASE("footer probes capture only the verified response validation tag", "[rest][validation_tag]") {