From dc203b6c25a4d05dd03589cb7589d593aaa09c02 Mon Sep 17 00:00:00 2001 From: Yu Date: Sun, 9 Aug 2026 08:52:28 +0800 Subject: [PATCH 1/8] test(io): extend loopback server for footer batches --- test/io/rest/loopback_range_server.hpp | 247 ++++++++++++++++++---- test/io/rest/test_rest_validation_tag.cpp | 185 ++++++++++++++++ 2 files changed, 387 insertions(+), 45 deletions(-) diff --git a/test/io/rest/loopback_range_server.hpp b/test/io/rest/loopback_range_server.hpp index 9912e26..1aa31d7 100644 --- a/test/io/rest/loopback_range_server.hpp +++ b/test/io/rest/loopback_range_server.hpp @@ -41,6 +41,7 @@ #include #include #include +#include #include #include @@ -76,12 +77,29 @@ 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; +}; + +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"); } @@ -131,10 +149,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 +197,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 +212,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,82 +227,111 @@ 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 get_idx = _get_count.fetch_add(1, std::memory_order_relaxed); + auto const key_idx = increment_request_count(_get_counts_by_key, key); + 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; @@ -279,19 +345,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 +400,61 @@ 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); } + } + + 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 +544,15 @@ 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}; + 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; 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]") { From dbd0b466ad32864c34231325b21ded85d225e2b2 Mon Sep 17 00:00:00 2001 From: Yu Date: Sun, 9 Aug 2026 09:06:33 +0800 Subject: [PATCH 2/8] test(io): define batched footer resolve conformance --- test/io/rest/test_rest_footer_resolve.cpp | 949 ++++++++++++++++++++++ 1 file changed, 949 insertions(+) create mode 100644 test/io/rest/test_rest_footer_resolve.cpp 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..d8f6090 --- /dev/null +++ b/test/io/rest/test_rest_footer_resolve.cpp @@ -0,0 +1,949 @@ +/* + * 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(); +} + +} // 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 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 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 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); + }); + + while (!first.ready_within(0ms)) { + REQUIRE(retained->wait_for_payload(1s)); + CHECK(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); + retained->release_one(); + REQUIRE(wait_until([&] { return server->get_count("second.parquet") == 1; }, 1s)); + + while (!second.ready_within(0ms)) { + REQUIRE(retained->wait_for_payload(1s)); + CHECK(retained->resident() <= budget); + if (second_delivered->load() == second_paths.size()) { break; } + retained->release_one(); + } + require_ready(second); + retained->release_all(); + + CHECK(errors->load() == 0); + CHECK(retained->resident() == 0); + CHECK(retained->peak() <= 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()); + auto fixture = make_ioctx(server, test_config(inflight)); + std::vector paths; + for (std::size_t i = 0; i < 12; ++i) { + paths.push_back(object_uri("reuse-" + std::to_string(i) + ".parquet")); + } + + 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); +} From 8abc92ff1ea4b5de34951e1952d6b9d3fecaddfc Mon Sep 17 00:00:00 2001 From: Yu Date: Sun, 9 Aug 2026 10:15:52 +0800 Subject: [PATCH 3/8] feat(io): batched footer resolve on rest_ioctx Resolving N objects' parquet footers costs N independent blocking suffix probes, each on a fresh TCP+TLS connection (the synchronous metadata path shares DNS/TLS-session state but not live connections). Add rest_ioctx::resolve_footer_objects(paths, on_result, stop): batched submission with streamed per-entry completion. One curl multi driven on the caller's thread carries every probe (and HEAD fallback) of a batch, reusing its pooled connections across entries, with per-entry semantics identical to open_io_object(path, parquet_footer_probe): verified-206 window, 200/416/unverifiable-206 HEAD fallback, the same retry policy per entry, per-attempt re-authorization and ETag capture, and the same perf-snapshot attribution. Each result carries a stashless io_object (size + validation tag) plus the footer window as a separate payload whose buffer is a lease on an ioctx-wide byte budget: bytes return when the buffer is freed, bounding resolve-ahead memory without attaching budget-held state to long-lived objects. While any transfer is active the engine acquires budget non-blockingly; a blocking, stop-aware wait happens only at zero active transfers. Batches FIFO-serialize per ioctx; a queued batch cancels out of the queue without side effects. Delivery is exactly-once per input occurrence on the caller's thread; cancellation aborts in-flight transfers and delivers one operation_canceled per undelivered entry; a throwing callback cancels the remainder and rethrows the first exception after the sweep. Config: footer_resolve_max_inflight (default derives n_reactors * max_connections; 0 disables the API) and footer_resolve_stash_budget (default 2 * inflight * footer_probe_bytes). admission_control gains try_acquire plus reserved/peak accessors, and the ioctx perf snapshot reports the budget's live/peak bytes. --- include/cucascade/exec/admission_control.hpp | 14 +- include/cucascade/io/rest/config.hpp | 20 + include/cucascade/io/rest/rest_ioctx.hpp | 50 ++- include/cucascade/io/rest/rest_reactor.hpp | 56 +++ src/exec/admission_control.cpp | 25 ++ src/io/rest/rest_ioctx.cpp | 159 ++++++++ src/io/rest/rest_reactor.cpp | 392 +++++++++++++++++++ test/CMakeLists.txt | 1 + 8 files changed, 715 insertions(+), 2 deletions(-) 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..5214a03 100644 --- a/include/cucascade/io/rest/config.hpp +++ b/include/cucascade/io/rest/config.hpp @@ -113,6 +113,26 @@ 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. + 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..e603698 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,30 @@ 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, or the API disabled via + /// @c config::footer_resolve_max_inflight == 0. 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 +145,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..28e18a7 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,157 @@ 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 == 0) { + 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)"); + } + + // 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) { + auto parsed = cucascade::io::parse(paths[i]); + if (parsed.scheme != "s3") { + parse_errors.emplace_back( + i, + std::make_exception_ptr(std::invalid_argument( + "rest_ioctx::resolve_footer_objects: unsupported scheme '" + parsed.scheme + "'"))); + continue; + } + 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); + } + + 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 +216,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..8542f67 100644 --- a/src/io/rest/rest_reactor.cpp +++ b/src/io/rest/rest_reactor.cpp @@ -1116,6 +1116,398 @@ 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; + } + + std::size_t undelivered = entries.size(); + std::size_t active = 0; + std::size_t next_to_start = 0; + std::exception_ptr callback_error; + + 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); + 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.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; + 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) { + e.lease = {}; + deliver(e, + error_result(e, + std::make_exception_ptr(std::runtime_error( + "rest_reactor::resolve_footer_batch: curl_easy_init failed")))); + return; + } + 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; + 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)); + 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; + }; + + 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; + } 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 + // lease is returned — a HEAD delivers no payload. + e.lease = {}; + e.kind = footer_entry_kind::head; + e.attempt = 0; + submit(e); + return; + } + if (status == 200 || status == 416) { + 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)) { + 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); + } + } + }; + + auto any_backoff = [&] { + return std::any_of(entries.begin(), entries.end(), [](footer_entry const& e) { + return e.stage == footer_entry_stage::backoff; + }); + }; + + // Returns false when a blocking budget wait was cut short by @p stop. + auto start_pending = [&] { + while (active < max_inflight && next_to_start < entries.size()) { + auto& e = entries[next_to_start]; + exec::admission_control::slot lease; + if (active > 0 || any_backoff()) { + // 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(); + for (auto& e : entries) { + if (active >= max_inflight) { break; } + if (e.stage == footer_entry_stage::backoff && e.retry_at <= now) { submit(e); } + } + }; + + auto poll_timeout_ms = [&] { + long timeout = 100; + auto const now = std::chrono::steady_clock::now(); + for (auto const& e : entries) { + if (e.stage != footer_entry_stage::backoff) { continue; } + auto const dt = + std::chrono::duration_cast(e.retry_at - now).count(); + timeout = std::min(timeout, std::max(1, static_cast(dt))); + } + return timeout; + }; + + 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)); + } + + 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 From 9270ed9702c1af2f37cf66df2d3504db2b363b86 Mon Sep 17 00:00:00 2001 From: Yu Date: Sun, 9 Aug 2026 10:15:53 +0800 Subject: [PATCH 4/8] test(io): size the reuse case to the payload lease and assert budget gauges The connection-reuse case retains all twelve footer payloads in its results while sizing the stash budget at four windows; under the payload-is-the-lease contract the resolve call then blocks waiting for bytes only its own caller could free. Size the budget to the retained payload count. The budget case now also asserts the snapshot gauges directly (reserved == peak == budget while payloads are held; reserved drops to zero with peak still at budget after release), and the scheduling-loop checks use fixed assertion counts so the suite's case and assertion totals stay stable. --- test/io/rest/test_rest_footer_resolve.cpp | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/test/io/rest/test_rest_footer_resolve.cpp b/test/io/rest/test_rest_footer_resolve.cpp index d8f6090..f54993c 100644 --- a/test/io/rest/test_rest_footer_resolve.cpp +++ b/test/io/rest/test_rest_footer_resolve.cpp @@ -676,9 +676,12 @@ TEST_CASE("batched footer resolve bounds retained payloads across concurrent bat ioctx->resolve_footer_objects(second_paths, second_callback); }); + bool stayed_within_budget = true; while (!first.ready_within(0ms)) { - REQUIRE(retained->wait_for_payload(1s)); - CHECK(retained->resident() <= budget); + 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(); } @@ -687,21 +690,30 @@ TEST_CASE("batched footer resolve bounds retained payloads across concurrent bat 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)) { - REQUIRE(retained->wait_for_payload(1s)); - CHECK(retained->resident() <= budget); + 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)); @@ -900,11 +912,11 @@ TEST_CASE("batched footer resolve reuses at most the configured in-flight connec { constexpr std::size_t inflight = 2; loopback_range_server server(test_payload()); - auto fixture = make_ioctx(server, test_config(inflight)); 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); From 0d84c614788db52212c146250aa613f9909337ae Mon Sep 17 00:00:00 2001 From: Yu Date: Sun, 9 Aug 2026 10:42:29 +0800 Subject: [PATCH 5/8] fix(io): harden batched footer resolve after review Memory now tracks the ledger: every fallback, terminal, and cancel path destroys the probe buffer before releasing its budget reservation, probe buffers reserve exactly the window up front so growth can never exceed the lease, and an explicit footer_resolve_stash_budget smaller than footer_probe_bytes is rejected at submission (a sub-window budget cannot be honored as a hard cap). Entry submission is exception-safe: an authorizer or curl setup failure becomes that entry's error and its siblings continue, and an unwind guard detaches any easy handle still attached to the batch multi before the owning entries are destroyed. Path parsing failures likewise become per-entry errors instead of aborting the batch with no callbacks. The event loop drops its full-vector scans for a backoff counter plus a deadline min-heap, giving O(N + R log N) scheduling instead of a worst-case O(N^2), and the per-submit clock read only happens when perf_instrumentation is on. --- include/cucascade/io/rest/config.hpp | 4 +- include/cucascade/io/rest/rest_ioctx.hpp | 8 +- src/io/rest/rest_ioctx.cpp | 30 +++-- src/io/rest/rest_reactor.cpp | 157 +++++++++++++++-------- 4 files changed, 131 insertions(+), 68 deletions(-) diff --git a/include/cucascade/io/rest/config.hpp b/include/cucascade/io/rest/config.hpp index 5214a03..8629770 100644 --- a/include/cucascade/io/rest/config.hpp +++ b/include/cucascade/io/rest/config.hpp @@ -131,7 +131,9 @@ struct config { /// 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. + /// 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}; }; diff --git a/include/cucascade/io/rest/rest_ioctx.hpp b/include/cucascade/io/rest/rest_ioctx.hpp index e603698..d03927e 100644 --- a/include/cucascade/io/rest/rest_ioctx.hpp +++ b/include/cucascade/io/rest/rest_ioctx.hpp @@ -116,9 +116,11 @@ class rest_ioctx : public templated_ioctx { /// @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, or the API disabled via - /// @c config::footer_resolve_max_inflight == 0. This ioctx must outlive - /// the call. + /// 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. + /// This ioctx must outlive the call. void resolve_footer_objects(std::span paths, std::function const& on_result, std::stop_token stop = {}); diff --git a/src/io/rest/rest_ioctx.cpp b/src/io/rest/rest_ioctx.cpp index 28e18a7..f011ef1 100644 --- a/src/io/rest/rest_ioctx.cpp +++ b/src/io/rest/rest_ioctx.cpp @@ -76,6 +76,13 @@ void rest_ioctx::resolve_footer_objects(std::span paths, 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. @@ -87,17 +94,20 @@ void rest_ioctx::resolve_footer_objects(std::span paths, valid_objects.reserve(paths.size()); valid_indices.reserve(paths.size()); for (std::size_t i = 0; i < paths.size(); ++i) { - auto parsed = cucascade::io::parse(paths[i]); - if (parsed.scheme != "s3") { - parse_errors.emplace_back( - i, - std::make_exception_ptr(std::invalid_argument( - "rest_ioctx::resolve_footer_objects: unsupported scheme '" + parsed.scheme + "'"))); - continue; + 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()); } - 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); } std::exception_ptr callback_error; diff --git a/src/io/rest/rest_reactor.cpp b/src/io/rest/rest_reactor.cpp index 8542f67..538ef15 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 @@ -1209,11 +1210,34 @@ void rest_reactor::resolve_footer_batch(std::span paths, entries[i].pos = i; } + // 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; @@ -1236,6 +1260,8 @@ void rest_reactor::resolve_footer_batch(std::span paths, 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, @@ -1253,6 +1279,7 @@ void rest_reactor::resolve_footer_batch(std::span paths, e.headers.reset(); if (e.stage == footer_entry_stage::transfer) { --active; } } + e.sink = suffix_sink{}; e.lease = {}; deliver(e, error_result(e, @@ -1264,45 +1291,59 @@ void rest_reactor::resolve_footer_batch(std::span paths, auto submit = [&](footer_entry& e) { bool const is_probe = e.kind == footer_entry_kind::probe; - 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) { + 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::make_exception_ptr(std::runtime_error( - "rest_reactor::resolve_footer_batch: curl_easy_init failed")))); - return; + deliver(e, error_result(e, std::current_exception())); } - 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; - 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)); - 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; }; auto schedule_retry = [&](footer_entry& e, std::string const& retry_after) { @@ -1319,6 +1360,8 @@ void rest_reactor::resolve_footer_batch(std::span paths, 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 + ")"); } @@ -1361,7 +1404,9 @@ void rest_reactor::resolve_footer_batch(std::span paths, return; } // Unverifiable 206: like the blocking path, fall back to a HEAD. The - // lease is returned — a HEAD delivers no payload. + // 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; @@ -1369,6 +1414,7 @@ void rest_reactor::resolve_footer_batch(std::span paths, return; } if (status == 200 || status == 416) { + e.sink = suffix_sink{}; e.lease = {}; e.kind = footer_entry_kind::head; e.attempt = 0; @@ -1438,18 +1484,12 @@ void rest_reactor::resolve_footer_batch(std::span paths, } }; - auto any_backoff = [&] { - return std::any_of(entries.begin(), entries.end(), [](footer_entry const& e) { - return e.stage == footer_entry_stage::backoff; - }); - }; - // Returns false when a blocking budget wait was cut short by @p stop. auto start_pending = [&] { while (active < max_inflight && next_to_start < entries.size()) { auto& e = entries[next_to_start]; exec::admission_control::slot lease; - if (active > 0 || any_backoff()) { + 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); @@ -1467,19 +1507,28 @@ void rest_reactor::resolve_footer_batch(std::span paths, auto resubmit_due = [&] { auto const now = std::chrono::steady_clock::now(); - for (auto& e : entries) { - if (active >= max_inflight) { break; } - if (e.stage == footer_entry_stage::backoff && e.retry_at <= now) { submit(e); } + while (!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; - auto const now = std::chrono::steady_clock::now(); - for (auto const& e : entries) { - if (e.stage != footer_entry_stage::backoff) { continue; } - auto const dt = - std::chrono::duration_cast(e.retry_at - now).count(); + 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; From cbf3dae2215a611fefa81f854693cd253dae2000 Mon Sep 17 00:00:00 2001 From: Yu Date: Sun, 9 Aug 2026 10:42:29 +0800 Subject: [PATCH 6/8] test(io): cover per-entry failure isolation and budget hardening Three cases on the batched footer resolve: multiple malformed-206 objects under a two-window budget complete alongside healthy siblings with the reservation gauge back at zero; an authorizer that throws for one key delivers that exception to that entry alone, with no GET issued for it and every sibling succeeding; and an unparsable URI in the batch is isolated as that entry's error, while an explicit stash budget smaller than the probe window is rejected at submission. The loopback harness gains per-key malformed-206 scripting and the mock authorizer gains per-key exception injection. --- test/io/rest/loopback_range_server.hpp | 20 ++++- test/io/rest/test_rest_footer_resolve.cpp | 98 +++++++++++++++++++++++ 2 files changed, 117 insertions(+), 1 deletion(-) diff --git a/test/io/rest/loopback_range_server.hpp b/test/io/rest/loopback_range_server.hpp index 1aa31d7..2fe3869 100644 --- a/test/io/rest/loopback_range_server.hpp +++ b/test/io/rest/loopback_range_server.hpp @@ -34,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -83,6 +84,7 @@ struct scripted_response { std::chrono::milliseconds delay{0}; std::optional etag; std::optional retry_after; + bool malformed_content_range{false}; }; struct key_response_script { @@ -337,7 +339,7 @@ class loopback_range_server { 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 { @@ -562,11 +564,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, {}}; } @@ -585,6 +601,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 index f54993c..5bdd444 100644 --- a/test/io/rest/test_rest_footer_resolve.cpp +++ b/test/io/rest/test_rest_footer_resolve.cpp @@ -306,6 +306,11 @@ rest_perf_snapshot snapshot_after_single_opens(loopback_range_server const& serv 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", @@ -400,6 +405,99 @@ TEST_CASE("batched footer resolve isolates per-object authorization and not-foun 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]") { From 12479c1c5087f93eb311d992fa20ad291def183c Mon Sep 17 00:00:00 2001 From: Yu Date: Sun, 9 Aug 2026 13:47:03 +0800 Subject: [PATCH 7/8] fix(io): enforce delivery contract on every batched-resolve exit path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After the first callback exception nothing may be delivered as success: completion draining, new-entry admission, and retry resubmission all stop at the first recorded throw (or a stop request), leaving undrained completions to the cancel sweep. A driver failure mid-batch (a curl multi error) now runs the same sweep before rethrowing, so every undelivered entry still receives exactly one canceled result on every exit path. An explicit footer_resolve_stash_budget of zero is no longer silently treated as the derived default — it falls under the sub-window rejection like any other too-small budget. And a zero footer_probe_bytes now matches the single-probe path exactly: batch entries skip the suffix GET and start at the HEAD fallback, with no lease taken. --- include/cucascade/io/rest/rest_ioctx.hpp | 5 +- src/io/rest/rest_ioctx.cpp | 4 +- src/io/rest/rest_reactor.cpp | 78 +++++++++++++++--------- 3 files changed, 54 insertions(+), 33 deletions(-) diff --git a/include/cucascade/io/rest/rest_ioctx.hpp b/include/cucascade/io/rest/rest_ioctx.hpp index d03927e..3f05bdd 100644 --- a/include/cucascade/io/rest/rest_ioctx.hpp +++ b/include/cucascade/io/rest/rest_ioctx.hpp @@ -120,7 +120,10 @@ class rest_ioctx : public templated_ioctx { /// @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. - /// This ioctx must outlive the call. + /// 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 = {}); diff --git a/src/io/rest/rest_ioctx.cpp b/src/io/rest/rest_ioctx.cpp index f011ef1..61782b3 100644 --- a/src/io/rest/rest_ioctx.cpp +++ b/src/io/rest/rest_ioctx.cpp @@ -44,9 +44,7 @@ rest_ioctx::rest_ioctx(std::size_t n_reactors, std::shared_ptr 0) { auto const& cfg = _reactors.front()->get_config(); std::size_t bytes = cfg.footer_resolve_stash_budget; - if (bytes == config::footer_resolve_auto || bytes == 0) { - bytes = 2 * inflight * cfg.footer_probe_bytes; - } + if (bytes == config::footer_resolve_auto) { bytes = 2 * inflight * cfg.footer_probe_bytes; } _footer_budget = std::make_shared(std::max(bytes, 1)); } } diff --git a/src/io/rest/rest_reactor.cpp b/src/io/rest/rest_reactor.cpp index 538ef15..0aa0b00 100644 --- a/src/io/rest/rest_reactor.cpp +++ b/src/io/rest/rest_reactor.cpp @@ -1208,6 +1208,10 @@ void rest_reactor::resolve_footer_batch(std::span paths, 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 @@ -1460,6 +1464,10 @@ void rest_reactor::resolve_footer_batch(std::span paths, 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; @@ -1486,19 +1494,22 @@ void rest_reactor::resolve_footer_batch(std::span paths, // Returns false when a blocking budget wait was cut short by @p stop. auto start_pending = [&] { - while (active < max_inflight && next_to_start < entries.size()) { + while (!callback_error && !stop.stop_requested() && active < max_inflight && + next_to_start < entries.size()) { auto& e = entries[next_to_start]; - 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; } + 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); } - e.lease = std::move(lease); ++next_to_start; submit(e); } @@ -1507,7 +1518,8 @@ void rest_reactor::resolve_footer_batch(std::span paths, auto resubmit_due = [&] { auto const now = std::chrono::steady_clock::now(); - while (!retry_heap.empty() && active < max_inflight) { + 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(); @@ -1534,24 +1546,32 @@ void rest_reactor::resolve_footer_batch(std::span paths, return timeout; }; - while (undelivered > 0) { - if (stop.stop_requested() || callback_error) { - cancel_remaining(); - break; - } - resubmit_due(); - if (!start_pending()) { - cancel_remaining(); - break; + 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)); } - 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); } From c0a26073ca918957e2a12eafa4afb7eb1a082367 Mon Sep 17 00:00:00 2001 From: Yu Date: Sun, 9 Aug 2026 13:47:04 +0800 Subject: [PATCH 8/8] test(io): pin the delivery contract under concurrency and zero-size knobs A deterministic GET barrier on the loopback server holds three requests and releases them together, so a callback that throws on the first delivery faces simultaneously-ready siblings: the case pins one success, every remaining entry canceled, only the initial three GETs on the wire, the original exception rethrown, and no callback after return. An explicit zero stash budget throws before any authorization or network request, falling under sub-window rejection rather than reading as the derived default. And a zero probe window matches the single-probe path: no GET and no reservation per entry, one HEAD each, with size and quoted ETag from the HEAD and a null footer. --- test/io/rest/loopback_range_server.hpp | 22 ++++- test/io/rest/test_rest_footer_resolve.cpp | 110 ++++++++++++++++++++++ 2 files changed, 130 insertions(+), 2 deletions(-) diff --git a/test/io/rest/loopback_range_server.hpp b/test/io/rest/loopback_range_server.hpp index 2fe3869..1aa752a 100644 --- a/test/io/rest/loopback_range_server.hpp +++ b/test/io/rest/loopback_range_server.hpp @@ -31,6 +31,7 @@ #include #include #include +#include #include #include #include @@ -56,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}; @@ -135,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); @@ -293,8 +297,9 @@ class loopback_range_server { 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); + 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, @@ -429,6 +434,17 @@ class loopback_range_server { 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) { @@ -552,6 +568,8 @@ class loopback_range_server { 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; diff --git a/test/io/rest/test_rest_footer_resolve.cpp b/test/io/rest/test_rest_footer_resolve.cpp index 5bdd444..2033656 100644 --- a/test/io/rest/test_rest_footer_resolve.cpp +++ b/test/io/rest/test_rest_footer_resolve.cpp @@ -374,6 +374,48 @@ TEST_CASE("batched footer resolve matches single-probe bytes, size, and validati } } +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]") { @@ -737,6 +779,74 @@ TEST_CASE( 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]") {