feat(http-client): add phase-specific timeouts

MODULE.bazel

@@ -1,6 +1,6 @@
 module(
     name = "khttpd",
-    version = "0.5.1",
+    version = "0.5.2",
 )
 
 bazel_dep(name = "platforms", version = "1.1.0")

doc/http-client.md

@@ -311,3 +311,9 @@ private:
 ws->set_header("Authorization", "Bearer token123");
 ws->connect("wss://api.example.com/ws", ...);
 ```
+
+### HTTP request deadlines
+
+`set_timeouts(ClientTimeouts{total, connect, read, write})` configures independent whole-request, DNS/TCP/TLS connection, response-read, and request-write deadlines. Values use `std::chrono::seconds`; zero disables the corresponding deadline. The earliest applicable deadline cancels the request with category `http.client.timeout` (1 total, 2 connect, 3 read, 4 write). The total deadline does not reset when data is transferred. `set_timeout(seconds)` applies the same limit to all four fields, including zero as unlimited. Stream reads and writes apply their corresponding phase limit to each operation.
+
+On-wire regression coverage is in `framework/tests/client_test.cpp`: delayed response with read/total limits, zero read/total, stalled TLS handshake, and a stalled upload. Gateway and provider-specific tests cover their streaming and proxy policies separately.

example/MODULE.bazel

@@ -8,7 +8,7 @@ bazel_dep(name = "boost", version = "1.90.0.bcr.1")
 bazel_dep(name = "boost.asio", version = "1.90.0.bcr.1")
 bazel_dep(name = "boost.mysql", version = "1.90.0.bcr.1")
 bazel_dep(name = "spdlog", version = "1.17.0")
-bazel_dep(name = "khttpd", version = "0.5.1")
+bazel_dep(name = "khttpd", version = "0.5.2")
 local_path_override(
     module_name = "khttpd",
     path = "..",

framework/client/client_timeouts.hpp

@@ -0,0 +1,35 @@
+#pragma once
+#include <chrono>
+#include <climits>
+#include <stdexcept>
+#include <boost/system/error_code.hpp>
+namespace khttpd::framework::client {
+struct ClientTimeouts {
+  std::chrono::seconds total{30}, connect{30}, read{30}, write{30};
+};
+inline void validate_client_timeouts(const ClientTimeouts& value) {
+  for (auto duration : {value.total, value.connect, value.read, value.write})
+    if (duration.count() < 0 || duration.count() > INT_MAX)
+      throw std::invalid_argument("client timeout must be between 0 and INT_MAX seconds");
+}
+class ClientTimeoutCategory final : public boost::system::error_category {
+ public:
+  const char* name() const noexcept override { return "http.client.timeout"; }
+  std::string message(int phase) const override {
+    switch(phase) {
+      case 1: return "HTTP client total timeout";
+      case 2: return "HTTP client connect timeout";
+      case 3: return "HTTP client read timeout";
+      case 4: return "HTTP client write timeout";
+      default: return "HTTP client timeout";
+    }
+  }
+  boost::system::error_condition default_error_condition(int) const noexcept override {
+    return boost::system::errc::make_error_condition(boost::system::errc::timed_out);
+  }
+};
+inline boost::system::error_code client_timeout_error(int phase) {
+  static ClientTimeoutCategory category;
+  return {phase,category};
+}
+}

framework/client/http_client.cpp

@@ -29,22 +29,44 @@ class Session : public std::enable_shared_from_this<Session> {
   http::request<http::string_body> req_;
   std::optional<http::response_parser<http::string_body>> response_parser_;
   beast::flat_buffer buffer_;
-  std::chrono::seconds timeout_;
+  ClientTimeouts timeouts_;
+  net::strand<net::io_context::executor_type> executor_;
+  net::steady_timer total_timer_, phase_timer_;
   std::atomic<bool> completed_{false};
 
  public:
-  Session(HttpClient::ResponseCallback callback, std::chrono::seconds timeout)
-      : callback_(std::move(callback)), timeout_(timeout) {}
+  Session(net::io_context& io, HttpClient::ResponseCallback callback, ClientTimeouts timeouts)
+      : callback_(std::move(callback)), timeouts_(timeouts), executor_(net::make_strand(io)),
+        total_timer_(executor_), phase_timer_(executor_) {}
 
   virtual ~Session() = default;
-  virtual void run(const std::string& host, const std::string& port,
-                   http::request<http::string_body> req) = 0;
+  void run(const std::string& host, const std::string& port, http::request<http::string_body> req) {
+    net::post(executor_, [self=shared_from_this(),host,port,req=std::move(req)]() mutable {
+      self->arm(self->total_timer_,self->timeouts_.total,1);
+      self->phase(self->timeouts_.connect,2);
+      self->start(host,port,std::move(req));
+    });
+  }
+  virtual void start(const std::string& host, const std::string& port,
+                     http::request<http::string_body> req) = 0;
   virtual void cancel() = 0;
 
  protected:
+  void arm(net::steady_timer& timer, std::chrono::seconds duration, int code) {
+    timer.cancel();
+    if (completed_.load() || duration.count()==0) return;
+    timer.expires_after(duration);
+    timer.async_wait([self=shared_from_this(),code](beast::error_code error) {
+      if (error || self->completed_.load()) return;
+      self->complete(client_timeout_error(code),{});
+      self->cancel();
+    });
+  }
+  void phase(std::chrono::seconds duration, int code) { arm(phase_timer_,duration,code); }
   void complete(beast::error_code ec, http::response<http::string_body> res) {
-    if (!completed_.exchange(true) && callback_) {
-      callback_(ec, std::move(res));
+    if (!completed_.exchange(true)) {
+      total_timer_.cancel(); phase_timer_.cancel();
+      if (callback_) callback_(ec, std::move(res));
     }
   }
 
@@ -69,15 +91,14 @@ class HttpSession : public Session {
 
  public:
   HttpSession(net::io_context& ioc, HttpClient::ResponseCallback cb,
-              std::chrono::seconds timeout)
-      : Session(std::move(cb), timeout),
-        stream_(net::make_strand(ioc)),
+              ClientTimeouts timeouts)
+      : Session(ioc, std::move(cb), timeouts),
+        stream_(executor_),
         resolver_(stream_.get_executor()) {}
 
-  void run(const std::string& host, const std::string& port,
+  void start(const std::string& host, const std::string& port,
            http::request<http::string_body> req) override {
     req_ = std::move(req);
-    stream_.expires_after(timeout_);
     resolver_.async_resolve(
         host, port,
         beast::bind_front_handler(&HttpSession::on_resolve, get_shared()));
@@ -94,29 +115,33 @@ class HttpSession : public Session {
   }
 
   void on_resolve(beast::error_code ec, tcp::resolver::results_type results) {
+    if (completed_.load()) return;
     if (ec) return on_fail(ec, "resolve");
-    stream_.expires_after(timeout_);
     stream_.async_connect(results, beast::bind_front_handler(
                                        &HttpSession::on_connect, get_shared()));
   }
 
   void on_connect(beast::error_code ec,
                   tcp::resolver::results_type::endpoint_type) {
+    if (completed_.load()) return;
     if (ec) return on_fail(ec, "connect");
-    stream_.expires_after(timeout_);
+    phase(timeouts_.write,4);
     http::async_write(
         stream_, req_,
         beast::bind_front_handler(&HttpSession::on_write, get_shared()));
   }
 
   void on_write(beast::error_code ec, std::size_t bytes_transferred) {
     boost::ignore_unused(bytes_transferred);
+    if (completed_.load()) return;
     if (ec) return on_fail(ec, "write");
+    phase(timeouts_.read,3);
 
     read_response();
   }
 
   void read_response() {
+    if (completed_.load()) return;
     response_parser_.emplace();
     response_parser_->body_limit((std::numeric_limits<std::uint64_t>::max)());
     response_parser_->skip(req_.method() == http::verb::head);
@@ -152,12 +177,12 @@ class HttpsSession : public Session {
 
  public:
   HttpsSession(net::io_context& ioc, ssl::context& ctx,
-               HttpClient::ResponseCallback cb, std::chrono::seconds timeout)
-      : Session(std::move(cb), timeout),
-        stream_(net::make_strand(ioc), ctx),
+               HttpClient::ResponseCallback cb, ClientTimeouts timeouts)
+      : Session(ioc, std::move(cb), timeouts),
+        stream_(executor_, ctx),
         resolver_(stream_.get_executor()) {}
 
-  void run(const std::string& host, const std::string& port,
+  void start(const std::string& host, const std::string& port,
            http::request<http::string_body> req) override {
     req_ = std::move(req);
     if (!SSL_set_tlsext_host_name(stream_.native_handle(), host.c_str())) {
@@ -167,7 +192,6 @@ class HttpsSession : public Session {
     }
     stream_.set_verify_callback(ssl::host_name_verification(host));
 
-    stream_.next_layer().expires_after(timeout_);
     resolver_.async_resolve(
         host, port,
         beast::bind_front_handler(&HttpsSession::on_resolve, get_shared()));
@@ -186,37 +210,41 @@ class HttpsSession : public Session {
   }
 
   void on_resolve(beast::error_code ec, tcp::resolver::results_type results) {
+    if (completed_.load()) return;
     if (ec) return on_fail(ec, "resolve");
-    stream_.next_layer().expires_after(timeout_);
     beast::get_lowest_layer(stream_).async_connect(
         results,
         beast::bind_front_handler(&HttpsSession::on_connect, get_shared()));
   }
 
   void on_connect(beast::error_code ec,
                   tcp::resolver::results_type::endpoint_type) {
+    if (completed_.load()) return;
     if (ec) return on_fail(ec, "connect");
-    stream_.next_layer().expires_after(timeout_);
     stream_.async_handshake(
         ssl::stream_base::client,
         beast::bind_front_handler(&HttpsSession::on_handshake, get_shared()));
   }
 
   void on_handshake(beast::error_code ec) {
+    if (completed_.load()) return;
     if (ec) return on_fail(ec, "handshake");
-    stream_.next_layer().expires_after(timeout_);
+    phase(timeouts_.write,4);
     http::async_write(
         stream_, req_,
         beast::bind_front_handler(&HttpsSession::on_write, get_shared()));
   }
 
   void on_write(beast::error_code ec, std::size_t bytes_transferred) {
     boost::ignore_unused(bytes_transferred);
+    if (completed_.load()) return;
     if (ec) return on_fail(ec, "write");
+    phase(timeouts_.read,3);
     read_response();
   }
 
   void read_response() {
+    if (completed_.load()) return;
     response_parser_.emplace();
     response_parser_->body_limit((std::numeric_limits<std::uint64_t>::max)());
     response_parser_->skip(req_.method() == http::verb::head);
@@ -320,7 +348,11 @@ void HttpClient::set_bearer_token(const std::string& token) {
 }
 
 void HttpClient::set_timeout(std::chrono::seconds seconds) {
-  timeout_ = seconds;
+  set_timeouts({seconds,seconds,seconds,seconds});
+}
+void HttpClient::set_timeouts(ClientTimeouts timeouts) {
+  validate_client_timeouts(timeouts);
+  timeouts_ = timeouts;
 }
 
 HttpClient::UrlParts HttpClient::parse_target(
@@ -421,10 +453,10 @@ void HttpClient::request(http::verb method, std::string path,
         return;
       }
       session = std::make_shared<HttpsSession>(ioc_, *ssl_ctx_ptr_,
-                                               std::move(callback), timeout_);
+                                               std::move(callback), timeouts_);
     } else {
       session =
-          std::make_shared<HttpSession>(ioc_, std::move(callback), timeout_);
+          std::make_shared<HttpSession>(ioc_, std::move(callback), timeouts_);
     }
     session->run(parts.host, parts.port, std::move(req));
   } catch (const std::exception& e) {
@@ -474,10 +506,10 @@ http::response<http::string_body> HttpClient::request_sync(
             beast::errc::operation_not_supported, beast::system_category()));
       }
       session = std::make_shared<HttpsSession>(ioc_, *ssl_ctx_ptr_,
-                                               std::move(callback), timeout_);
+                                               std::move(callback), timeouts_);
     } else {
       session =
-          std::make_shared<HttpSession>(ioc_, std::move(callback), timeout_);
+          std::make_shared<HttpSession>(ioc_, std::move(callback), timeouts_);
     }
     session->run(parts.host, parts.port, std::move(req));
   } catch (const boost::system::system_error&) {
@@ -487,12 +519,13 @@ http::response<http::string_body> HttpClient::request_sync(
         beast::errc::invalid_argument, beast::system_category()));
   }
 
-  if (f.wait_for(timeout_ + std::chrono::seconds(1)) !=
+  if (timeouts_.total.count() > 0 &&
+      f.wait_for(timeouts_.total + std::chrono::seconds(1)) !=
       std::future_status::ready) {
     completed->store(true);
     if (session) session->cancel();
     throw boost::system::system_error(
-        beast::error_code(beast::errc::timed_out, beast::system_category()));
+        client_timeout_error(1));
   }
   auto result = f.get();
 

framework/client/http_client.hpp

@@ -11,6 +11,7 @@
 #include <boost/json.hpp>
 #include <boost/url.hpp>
 #include <functional>
+#include "client_timeouts.hpp"
 #include <future>
 #include <map>
 #include <memory>
@@ -92,6 +93,7 @@ class HttpClient : public std::enable_shared_from_this<HttpClient> {
   void set_default_header(const std::string& key, const std::string& value);
   void set_bearer_token(const std::string& token);
   void set_timeout(std::chrono::seconds seconds);
+  void set_timeouts(ClientTimeouts timeouts);
 
   // Core Request Method (Used by Macros)
   void request(http::verb method,
@@ -130,7 +132,7 @@ class HttpClient : public std::enable_shared_from_this<HttpClient> {
   std::unique_ptr<HostPool>
       host_pool_;  // Multi-host support (null if single host)
   std::map<std::string, std::string> default_headers_;
-  std::chrono::seconds timeout_{30};
+  ClientTimeouts timeouts_;
 };
 }  // namespace khttpd::framework::client
 

framework/client/http_client_stream.cpp

@@ -44,6 +44,7 @@ struct HttpClientStream::Impl
   http::verb request_method = http::verb::unknown;
   std::string host;
   std::string port;
+  std::function<void()> start_write_callback;
   bool use_tls = false;
   bool started = false;
   bool request_finished = false;
@@ -137,6 +138,7 @@ struct HttpClientStream::Impl
 
   template <class Stream>
   void async_write_header(Stream& stream, Callback callback) {
+    if (start_write_callback) start_write_callback();
     http::async_write_header(
         stream, *request_serializer,
         [self = shared_from_this(), callback = std::move(callback)](
@@ -321,4 +323,7 @@ void HttpClientStream::async_read_some(net::mutable_buffer buffer,
   impl_->read(buffer, std::move(cb));
 }
 void HttpClientStream::cancel() { impl_->cancel(); }
+void HttpClientStream::set_start_write_callback(std::function<void()> callback) {
+  impl_->start_write_callback=std::move(callback);
+}
 }  // namespace khttpd::framework::client

framework/client/http_client_stream.hpp

@@ -34,6 +34,8 @@ class HttpClientStream : public std::enable_shared_from_this<HttpClientStream> {
   void async_read_some(boost::asio::mutable_buffer buffer,
                        ReadCallback callback);
   void cancel();
+  // Set before async_start; called on the stream executor before sending the head.
+  void set_start_write_callback(std::function<void()> callback);
 
  private:
   struct Impl;