core: net/http_client — libcurl multi wrapper (stage 2)

One HttpClient owns a small pool of workers, each with its own CURLM; an easy handle lives on one worker for its life. Public start/pause/resume/ cancel enqueue a command + curl_multi_wakeup(); callbacks (on_head / on_data / on_finished) run on the worker thread and return a DataAction (proceed / pause / abort). Covers redirects (final-response head only), ranges (inclusive ByteRange -> CURLOPT_RANGE), proxy/SOCKS5, basic/digest auth, cookies, verbatim headers, stall detection, a coarse recv-rate cap, and a curl_share DNS/TLS cache across workers. CURLcode + HTTP status -> vdm::Error in net/curl_error. A probe is on_head returning abort: it finishes successfully (head_complete), not canceled. Tests drive tools/testserver: full GET, ranged 206, redirect chain, 404 -> not_found, connection refused -> connect_failed, HEAD probe + ranged 0-0 probe (no body), cancel mid-transfer, pause/resume completes. Skip cleanly if testserver isn't in the tree. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS
2026-09-09 23:31:09 +04:00
parent e442c99130
commit bd1bc029f3
9 changed files with 1369 additions and 4 deletions
@@ -0,0 +1,270 @@
+#include "vdm/net/http_client.hpp"
+
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "testserver_fixture.hpp"
+#include "vtest.hpp"
+
+using namespace vdm;
+using namespace vdm::net;
+using vdm::testing::TestServer;
+
+namespace {
+
+// Collects callback output from a transfer and lets the test thread wait for the end.
+struct Recorder {
+    std::mutex mu;
+    ResponseHead head;
+    bool head_seen = false;
+    std::uint64_t bytes = 0;
+    std::promise<Result<TransferStats>> done;
+    std::future<Result<TransferStats>> done_fut = done.get_future();
+
+    DataAction want_on_head = DataAction::proceed;  // set before start()
+    std::atomic<DataAction> want_on_data{DataAction::proceed};
+    std::atomic<int> data_calls{0};
+
+    TransferCallbacks callbacks() {
+        return TransferCallbacks{
+            .on_head =
+                [this](const ResponseHead &h) {
+                    std::lock_guard lk(mu);
+                    head = h;
+                    head_seen = true;
+                    return want_on_head;
+                },
+            .on_data =
+                [this](ConstByteSpan s) {
+                    data_calls.fetch_add(1);
+                    std::lock_guard lk(mu);
+                    bytes += s.size();
+                    return want_on_data.load();
+                },
+            .on_finished = [this](Result<TransferStats> r) { done.set_value(std::move(r)); },
+        };
+    }
+
+    Result<TransferStats> wait(std::chrono::seconds to = std::chrono::seconds(20)) {
+        if (done_fut.wait_for(to) != std::future_status::ready)
+            return Err{Error::timeout, "test wait timed out"};
+        return done_fut.get();
+    }
+};
+
+// on_data needs an atomic for the cancel/pause tests to flip it from the test thread.
+struct AtomicAction {
+    std::atomic<DataAction> a{DataAction::proceed};
+    void store(DataAction v) { a.store(v); }
+    operator DataAction() const { return a.load(); }
+    DataAction load() const { return a.load(); }
+};
+
+}  // namespace
+
+VT_TEST(http_plain_full_get) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    HttpClient client({.workers = 1});
+
+    Recorder rec;
+    Request req;
+    req.url = srv.url("/plain/file/64K");
+    auto t = client.start(std::move(req), rec.callbacks());
+
+    auto r = rec.wait();
+    VT_REQUIRE(r.has_value());
+    VT_CHECK_EQ(r.value().http_status, 200);
+    VT_CHECK_EQ(r.value().bytes_received, 65536u);
+    VT_CHECK(rec.head_seen);
+    VT_CHECK_EQ(rec.head.status, 200);
+    VT_CHECK(rec.head.headers.has("Accept-Ranges"));
+    VT_CHECK_EQ(rec.bytes, 65536u);
+    VT_CHECK(t.id() != 0);
+}
+
+VT_TEST(http_ranged_get_is_206) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    HttpClient client({.workers = 1});
+
+    Recorder rec;
+    Request req;
+    req.url = srv.url("/plain/file/64K");
+    req.range = ByteRange{1000, 1999};
+    auto t = client.start(std::move(req), rec.callbacks());
+
+    auto r = rec.wait();
+    VT_REQUIRE(r.has_value());
+    VT_CHECK_EQ(rec.head.status, 206);
+    auto cr = rec.head.headers.get("Content-Range");
+    VT_REQUIRE(cr.has_value());
+    VT_CHECK_EQ(std::string(*cr), std::string("bytes 1000-1999/65536"));
+    VT_CHECK_EQ(rec.bytes, 1000u);
+    VT_CHECK_EQ(r.value().bytes_received, 1000u);
+}
+
+VT_TEST(http_follows_redirect_chain) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    HttpClient client({.workers = 1});
+
+    Recorder rec;
+    Request req;
+    req.url = srv.url("/redirect-chain/file/16K");
+    auto t = client.start(std::move(req), rec.callbacks());
+
+    auto r = rec.wait();
+    VT_REQUIRE(r.has_value());
+    VT_CHECK_EQ(rec.head.status, 200);  // on_head fires once, for the final response
+    VT_CHECK_EQ(rec.bytes, 16u * 1024u);
+    VT_CHECK(r.value().effective_url != srv.url("/redirect-chain/file/16K"));
+    VT_CHECK(r.value().effective_url.find("_r=done") != std::string::npos);
+}
+
+VT_TEST(http_404_is_not_found_error) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    HttpClient client({.workers = 1});
+
+    Recorder rec;
+    Request req;
+    req.url = srv.url("/plain/nope");
+    auto t = client.start(std::move(req), rec.callbacks());
+
+    auto r = rec.wait();
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.error().code, Error::not_found);
+    VT_CHECK_EQ(r.error().http_status, 404);
+}
+
+VT_TEST(http_connection_refused_is_connect_failed) {
+    HttpClient client({.workers = 1});
+    Recorder rec;
+    Request req;
+    req.url = "http://127.0.0.1:1/nothing";  // nothing listens on :1
+    req.connect_timeout_ms = 2000;
+    auto t = client.start(std::move(req), rec.callbacks());
+
+    auto r = rec.wait();
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.error().code, Error::connect_failed);
+}
+
+VT_TEST(http_head_probe_stops_after_headers) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    HttpClient client({.workers = 1});
+
+    Recorder rec;
+    rec.want_on_head = DataAction::abort;  // probe: headers only
+    Request req;
+    req.url = srv.url("/plain/file/1M");
+    req.method = Method::head;
+    auto t = client.start(std::move(req), rec.callbacks());
+
+    auto r = rec.wait();
+    VT_REQUIRE(r.has_value());  // head_complete, NOT canceled
+    VT_CHECK(rec.head_seen);
+    VT_REQUIRE(rec.head.content_length.has_value());
+    VT_CHECK_EQ(*rec.head.content_length, 1024u * 1024u);
+    VT_CHECK_EQ(rec.data_calls.load(), 0);
+}
+
+VT_TEST(http_ranged_get_probe_stops_without_downloading_file) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    HttpClient client({.workers = 1});
+
+    Recorder rec;
+    rec.want_on_head = DataAction::abort;
+    Request req;
+    req.url = srv.url("/plain/file/8M");
+    req.range = ByteRange{0, 0};  // classic HEAD-refused fallback
+    auto t = client.start(std::move(req), rec.callbacks());
+
+    auto r = rec.wait();
+    VT_REQUIRE(r.has_value());
+    VT_CHECK(rec.bytes <= 1u);  // at most the one probe byte, usually 0
+}
+
+VT_TEST(http_cancel_mid_transfer) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    HttpClient client({.workers = 1});
+
+    AtomicAction data_action;
+    std::promise<Result<TransferStats>> done;
+    auto fut = done.get_future();
+    std::atomic<int> calls{0};
+
+    TransferCallbacks cbs{
+        .on_head = [](const ResponseHead &) { return DataAction::proceed; },
+        .on_data =
+            [&](ConstByteSpan) {
+                calls.fetch_add(1);
+                return data_action.load();
+            },
+        .on_finished = [&](Result<TransferStats> r) { done.set_value(std::move(r)); },
+    };
+
+    Request req;
+    req.url = srv.url("/throttled/file/4M");  // ~128 KiB/s => lots of chunks
+    auto t = client.start(std::move(req), std::move(cbs));
+
+    // wait for the transfer to actually start, then cancel
+    for (int i = 0; i < 200 && calls.load() == 0; ++i)
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    VT_REQUIRE(calls.load() > 0);
+    t.cancel();
+
+    VT_REQUIRE(fut.wait_for(std::chrono::seconds(10)) == std::future_status::ready);
+    auto r = fut.get();
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.error().code, Error::canceled);
+}
+
+VT_TEST(http_pause_then_resume_completes) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    HttpClient client({.workers = 1});
+
+    std::atomic<int> calls{0};
+    std::atomic<std::uint64_t> total{0};
+    std::promise<Result<TransferStats>> done;
+    auto fut = done.get_future();
+
+    TransferCallbacks cbs{
+        .on_head = [](const ResponseHead &) { return DataAction::proceed; },
+        .on_data =
+            [&](ConstByteSpan s) {
+                calls.fetch_add(1);
+                total.fetch_add(s.size());
+                return DataAction::proceed;
+            },
+        .on_finished = [&](Result<TransferStats> r) { done.set_value(std::move(r)); },
+    };
+
+    Request req;
+    req.url = srv.url("/throttled/file/1M");
+    auto t = client.start(std::move(req), std::move(cbs));
+
+    for (int i = 0; i < 200 && calls.load() == 0; ++i)
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    VT_REQUIRE(calls.load() > 0);
+
+    t.pause();
+    int calls_at_pause = calls.load();
+    std::this_thread::sleep_for(std::chrono::milliseconds(400));
+    VT_CHECK(calls.load() - calls_at_pause <= 1);  // at most one in-flight chunk slips through
+    t.resume();
+
+    VT_REQUIRE(fut.wait_for(std::chrono::seconds(20)) == std::future_status::ready);
+    auto r = fut.get();
+    VT_REQUIRE(r.has_value());
+    VT_CHECK_EQ(total.load(), 1024u * 1024u);
+}
@@ -0,0 +1,113 @@
+// testserver_fixture.hpp — spawn tools/testserver for a test, tear it down after.
+//
+// Linux-only (fork/exec/pipe/kill). The path to testserver.py is injected by CMake as
+// VDM_TESTSERVER_PY; if it's empty or missing the fixture reports unavailable() and the
+// test should skip.
+
+#ifndef VDM_TESTS_NET_TESTSERVER_FIXTURE_HPP
+#define VDM_TESTS_NET_TESTSERVER_FIXTURE_HPP
+
+#include <fcntl.h>
+#include <signal.h>
+#include <sys/wait.h>
+#include <unistd.h>
+
+#include <cerrno>
+#include <chrono>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <string>
+#include <thread>
+
+#ifndef VDM_TESTSERVER_PY
+#define VDM_TESTSERVER_PY ""
+#endif
+
+namespace vdm::testing {
+
+class TestServer {
+  public:
+    TestServer() {
+        const char *script = VDM_TESTSERVER_PY;
+        if (!script || !*script || ::access(script, R_OK) != 0)
+            return;
+
+        int pipefd[2];
+        if (::pipe(pipefd) != 0)
+            return;
+
+        pid_ = ::fork();
+        if (pid_ < 0) {
+            ::close(pipefd[0]);
+            ::close(pipefd[1]);
+            return;
+        }
+        if (pid_ == 0) {
+            ::dup2(pipefd[1], STDOUT_FILENO);
+            ::close(pipefd[0]);
+            ::close(pipefd[1]);
+            int devnull = ::open("/dev/null", O_WRONLY);
+            if (devnull >= 0)
+                ::dup2(devnull, STDERR_FILENO);
+            ::execlp("python3", "python3", script, "--port", "0", "--seed", "9", "--loris-seconds",
+                     "1", "--throttle-bps", "131072", static_cast<char *>(nullptr));
+            ::_exit(127);
+        }
+        ::close(pipefd[1]);
+
+        // Read the port line the server prints to stdout.
+        std::string line;
+        char c = 0;
+        auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
+        while (std::chrono::steady_clock::now() < deadline) {
+            ssize_t r = ::read(pipefd[0], &c, 1);
+            if (r == 1) {
+                if (c == '\n')
+                    break;
+                line += c;
+            } else if (r == 0) {
+                break;
+            } else if (errno != EINTR) {
+                break;
+            }
+        }
+        ::close(pipefd[0]);
+        if (!line.empty())
+            port_ = std::atoi(line.c_str());
+
+        // Give the listener a moment to accept.
+        std::this_thread::sleep_for(std::chrono::milliseconds(150));
+    }
+
+    ~TestServer() {
+        if (pid_ > 0) {
+            ::kill(pid_, SIGTERM);
+            int status = 0;
+            for (int i = 0; i < 50; ++i) {
+                if (::waitpid(pid_, &status, WNOHANG) == pid_)
+                    return;
+                std::this_thread::sleep_for(std::chrono::milliseconds(20));
+            }
+            ::kill(pid_, SIGKILL);
+            ::waitpid(pid_, &status, 0);
+        }
+    }
+
+    TestServer(const TestServer &) = delete;
+    TestServer &operator=(const TestServer &) = delete;
+
+    [[nodiscard]] bool available() const { return port_ > 0; }
+    [[nodiscard]] int port() const { return port_; }
+    [[nodiscard]] std::string url(const std::string &path) const {
+        return "http://127.0.0.1:" + std::to_string(port_) + path;
+    }
+
+  private:
+    pid_t pid_ = -1;
+    int port_ = 0;
+};
+
+}  // namespace vdm::testing
+
+#endif  // VDM_TESTS_NET_TESTSERVER_FIXTURE_HPP