feat(http): let another thread cancel a request in flight

There was no way to abandon a request that had been sent. A caller that stopped
caring (a user pressing Esc, say) could only leave the call running on a thread
and wait out readTimeoutMs, with the connection held open and the server still
producing the response.

send and send_stream take an optional std::stop_token. While a token is
attached, every wait on the socket is made in slices of at most 50 ms with a
check for a stop request between them, and bio_recv waits for the socket before
reading, so a handshake or the second half of a TLS record is covered too. When
the token is stopped the call returns within about one slice, and the connection
is dropped rather than pooled, so the server sees it end. A close_notify goes
out only when the stop comes after the handshake has finished.

HttpResponse::cancelled says why the call ended, without matching text. Before a
status line has arrived the response is statusCode 0 with statusText and
bodyError "Cancelled"; after it, statusCode is the server's and bodyError is
"cancelled". A cancelled request is not resent on a fresh connection by the
stale-connection retry, and a redirect is not followed.

proxy_tunnel and proxy_connect take the token as a trailing default argument
and hand it to the socket or the TLS session they open, so the exchange with
an HTTP, https or SOCKS5 proxy is covered as well.

Without a token (stop_possible() is false) every wait is the single poll it was,
and bio_recv does not poll first.

Not covered: getaddrinfo, a write blocked because the peer is not reading, and
connect on openkal, where connect completes synchronously. download_to_file is
unchanged.

Tests in test_cancel run against the in-process TLS server and a listener that
accepts and says nothing: stopped while waiting for the response head, while a
stream waits for its next chunk, and during the handshake, each with a 30 s read
timeout so only the token can have ended it; a token already stopped sends
nothing; a token never stopped leaves a 300 ms timeout alone; a cancelled
connection is not reused; a cancelled request on a pooled connection is not
retried; and 20 cancelled requests leave the descriptor count where it was.

Not run on macOS or Windows; the change adds no platform-specific code beyond
the poll the sockets already use.

Author: yspbwx2010

CHANGELOG.md

@@ -1,5 +1,23 @@
 # Changelog
 
+## Unreleased
+
+`send` and `send_stream` take an optional `std::stop_token`, so another thread
+can abandon a request that is in flight.
+
+* When the token is stopped the call returns within about 50 ms with
+  `HttpResponse::cancelled` set, and the connection is closed instead of being
+  returned to the pool. It covers connect, the proxy CONNECT exchange, the TLS
+  handshake, waiting for the response head, reading the body and the gaps
+  between streaming callbacks.
+* Before the status line arrives the response is `statusCode` 0 with
+  `statusText` and `bodyError` `Cancelled`. After it, `statusCode` is the
+  server's and `bodyError` is `cancelled`. A cancelled request is not retried on
+  a new connection and its redirect is not followed.
+* Not covered: name resolution, and a write that is blocked because the server
+  is not reading. Without a token nothing changes.
+* `proxy_connect` takes the token as a trailing default argument.
+
 ## 0.3.3
 
 **Behaviour change, and the reason for this release:** `verifySsl = true` (the

README.md

@@ -46,6 +46,24 @@ else                       { /* resp.body is all of it */ }
 `ok()` deliberately does not consult `bodyComplete`, so existing `if (res.ok())`
 means exactly what it did before.
 
+### Cancelling a request
+
+Pass a `std::stop_token` to `send` or `send_stream` and stop its source from
+another thread. The call returns within about 50 ms with `cancelled` set, and
+the connection is closed rather than pooled.
+
+```cpp
+std::stop_source source;
+auto worker = std::jthread([&] { response = client.send(request, source.get_token()); });
+// elsewhere:
+source.request_stop();
+```
+
+If the status line had not arrived, `statusCode` is 0 and `statusText` is
+`Cancelled`; otherwise they are the server's and `bodyError` is `cancelled`.
+Name resolution and a write blocked on a server that is not reading cannot be
+interrupted.
+
 ### Configuration
 
 | field | default | what it decides |

src/http.cppm

@@ -58,6 +58,11 @@ export struct HttpResponse {
     // "Invalid chunk size: zz" where the server had said "OK".
     std::string bodyError;
 
+    // True when the caller's stop token ended the request. Then `bodyComplete`
+    // is false, and the status line is the server's if it had arrived and
+    // 0 / "Cancelled" if not.
+    bool cancelled { false };
+
     bool ok() const { return statusCode >= 200 && statusCode < 300; }
 };
 
@@ -880,14 +885,21 @@ public:
     HttpClient(HttpClient&&) = default;
     HttpClient& operator=(HttpClient&&) = default;
 
-    HttpResponse send(const HttpRequest& request) {
-        return send_impl(request, 0);
+    // `stop` can be stopped from another thread to abandon the request: the
+    // call returns within about 50 ms with `cancelled` set, and the connection
+    // is closed rather than pooled. It cannot interrupt name resolution or a
+    // write blocked because the server is not reading.
+    HttpResponse send(const HttpRequest& request, std::stop_token stop = {}) {
+        return send_impl(request, 0, stop);
     }
 
     // Streaming SSE request — reads the response body incrementally, feeding
     // chunks through SseParser to the caller's callback. The callback receives
     // each SseEvent and returns true to continue or false to stop.
-    HttpResponse send_stream(const HttpRequest& request, SseCallbackFn callback) {
+    // `stop` is as for send().
+    HttpResponse send_stream(const HttpRequest& request, SseCallbackFn callback,
+                             std::stop_token stop = {}) {
+        if (stop.stop_requested()) return cancelled_response();
         HttpResponse response;
 
         auto parsed = parse_url(request.url);
@@ -901,12 +913,13 @@ public:
         PooledConnection guard(pool_, poolKey);
 
         auto exchange = perform_exchange(parsed, poolKey,
-                                         build_request(request, parsed), guard);
+                                         build_request(request, parsed), guard, stop);
         if (!exchange.error.empty()) {
             response.statusCode = 0;
             response.statusText = exchange.error;
             response.bodyComplete = false;
             response.bodyError = exchange.error;
+            response.cancelled = exchange.cancelled;
             return response;
         }
 
@@ -921,6 +934,7 @@ public:
         SseParser parser;
 
         auto sink = [&](std::string_view data) -> bool {
+            if (stop.stop_requested()) return false;
             if (captureBody) {
                 append_within_limit(response.body, data, stream_error_body_limit);
             }
@@ -932,7 +946,7 @@ public:
 
         finish_body(response, exchange, request.method,
                     /*maxBytes=*/std::numeric_limits<std::int64_t>::max(),
-                    sink, guard);
+                    sink, guard, stop);
         return response;
     }
 
@@ -959,8 +973,18 @@ private:
         TlsSocket* sock { nullptr };
         ResponseHead head;
         std::string error;
+        bool cancelled { false };
     };
 
+    static HttpResponse cancelled_response() {
+        HttpResponse response;
+        response.statusText = "Cancelled";
+        response.bodyComplete = false;
+        response.bodyError = "Cancelled";
+        response.cancelled = true;
+        return response;
+    }
+
     std::string build_request(const HttpRequest& request, const ParsedUrl& parsed) const {
         std::string reqStr;
         reqStr += method_to_string(request.method);
@@ -999,11 +1023,12 @@ private:
 
     // False on failure, with `error` set when there is more to say than that the
     // connection failed: a proxy that refused the tunnel says so in its own words.
-    bool open_connection(TlsSocket& sock, const ParsedUrl& parsed, std::string& error) {
+    bool open_connection(TlsSocket& sock, const ParsedUrl& parsed, std::string& error,
+                         std::stop_token stop) {
         if (config_.proxy.has_value()) {
             auto tunnel = proxy_tunnel(parse_proxy_url(config_.proxy.value()),
                                        parsed.host, parsed.port,
-                                       config_.connectTimeoutMs, config_.verifySsl);
+                                       config_.connectTimeoutMs, config_.verifySsl, stop);
             if (!tunnel.ok()) {
                 error = std::move(tunnel.error);
                 return false;
@@ -1028,8 +1053,17 @@ private:
     // request, so sending it again on a fresh connection is safe — once, and
     // only when no response byte has arrived. See `retryOnStaleConnection`.
     Exchange perform_exchange(const ParsedUrl& parsed, const std::string& poolKey,
-                              const std::string& reqStr, PooledConnection& guard) {
+                              const std::string& reqStr, PooledConnection& guard,
+                              std::stop_token stop = {}) {
+        // Drops the connection; a stop request outranks whatever went wrong.
+        auto fail = [&](std::string message) -> Exchange {
+            guard.drop();
+            if (stop.stop_requested()) return { nullptr, {}, "Cancelled", true };
+            return { nullptr, {}, std::move(message) };
+        };
+
         for (int attempt = 0; attempt < 2; ++attempt) {
+            if (stop.stop_requested()) return fail("");
             bool reused = false;
             TlsSocket* sock = nullptr;
 
@@ -1041,15 +1075,17 @@ private:
                 if (it != pool_.end()) pool_.erase(it);
                 auto [inserted, ok] = pool_.emplace(poolKey, TlsSocket{});
                 sock = &inserted->second;
+            }
+            sock->set_stop(stop);
+            if (!reused) {
                 std::string openError;
-                if (!open_connection(*sock, parsed, openError)) {
+                if (!open_connection(*sock, parsed, openError, stop)) {
                     // The proxy's refusal, else the TLS session's reason,
                     // else the TCP connection failed and there is no more to say.
                     std::string why = std::move(openError);
                     if (why.empty()) why = sock->error();
                     if (why.empty()) why = "Connection failed";
-                    guard.drop();
-                    return { nullptr, {}, std::move(why) };
+                    return fail(std::move(why));
                 }
             }
 
@@ -1065,8 +1101,7 @@ private:
             // connection therefore yields exactly one execution.
             if (!write_all(*sock, reqStr, config_.readTimeoutMs)) {
                 if (mayRetry) { guard.reset(); continue; }
-                guard.drop();
-                return { nullptr, {}, "Write failed" };
+                return fail("Write failed");
             }
 
             auto head = read_response_head(*sock, config_.readTimeoutMs);
@@ -1076,18 +1111,16 @@ private:
             // arrived the server has seen the request, and repeating it could
             // repeat its effect.
             if (mayRetry && !head.error().sawBytes) { guard.reset(); continue; }
-            guard.drop();
-            return { nullptr, {}, head.error().message };
+            return fail(head.error().message);
         }
-        guard.drop();
-        return { nullptr, {}, "No response" };
+        return fail("No response");
     }
 
     // Reads the body into `sink` and settles the pool guard by what the read
     // found. The single place that decides whether a connection is reusable.
     void finish_body(HttpResponse& response, Exchange& exchange, Method method,
                      std::int64_t maxBytes, const BodySink& sink,
-                     PooledConnection& guard) {
+                     PooledConnection& guard, std::stop_token stop = {}) {
         if (!response_has_body(method, exchange.head.statusCode)) {
             if (config_.keepAlive && !exchange.head.connectionClose) guard.keep();
             return;
@@ -1103,18 +1136,28 @@ private:
                 // Not an error, but bytes are still owed on the socket.
                 response.bodyComplete = false;
                 response.bodyError = "stopped by callback";
+                if (stop.stop_requested()) {
+                    response.bodyError = "cancelled";
+                    response.cancelled = true;
+                }
                 break;
             case BodyEnd::ClosedByPeer:
                 // The body ended where it said it would; the connection did too.
                 break;
             case BodyEnd::Truncated:
                 response.bodyComplete = false;
                 response.bodyError = outcome.error;
+                if (stop.stop_requested()) {
+                    response.bodyError = "cancelled";
+                    response.cancelled = true;
+                }
                 break;
         }
     }
 
-    HttpResponse send_impl(const HttpRequest& request, int redirectCount) {
+    HttpResponse send_impl(const HttpRequest& request, int redirectCount,
+                           std::stop_token stop) {
+        if (stop.stop_requested()) return cancelled_response();
         HttpResponse response;
 
         auto parsed = parse_url(request.url);
@@ -1128,12 +1171,13 @@ private:
         PooledConnection guard(pool_, poolKey);
 
         auto exchange = perform_exchange(parsed, poolKey,
-                                         build_request(request, parsed), guard);
+                                         build_request(request, parsed), guard, stop);
         if (!exchange.error.empty()) {
             response.statusCode = 0;
             response.statusText = exchange.error;
             response.bodyComplete = false;
             response.bodyError = exchange.error;
+            response.cancelled = exchange.cancelled;
             return response;
         }
 
@@ -1157,10 +1201,10 @@ private:
                         response.body.append(data);
                         return true;
                     },
-                    guard);
+                    guard, stop);
 
         // Follow 3xx redirects if configured.
-        if (config_.maxRedirects > 0 &&
+        if (config_.maxRedirects > 0 && !response.cancelled &&
             response.statusCode >= 300 && response.statusCode < 400 &&
             redirectCount < config_.maxRedirects) {
             std::string location = find_header(response.headers, "Location");
@@ -1181,7 +1225,7 @@ private:
                 // connection into the pool under this same key, and a guard
                 // still armed would delete it when this scope ends.
                 guard.drop();
-                return send_impl(redirectReq, redirectCount + 1);
+                return send_impl(redirectReq, redirectCount + 1, stop);
             }
         }
 

src/proxy.cppm

@@ -351,8 +351,10 @@ export struct ProxyTunnel {
 // setting that governs the connection to the target.
 export ProxyTunnel proxy_tunnel(const ProxyConfig& proxy,
                                 std::string_view targetHost, int targetPort,
-                                int timeoutMs, bool verifySsl = true) {
+                                int timeoutMs, bool verifySsl = true,
+                                std::stop_token stop = {}) {
     ProxyTunnel tunnel;
+    tunnel.socket.set_stop(stop);
     const std::string where = proxy.host + ":" + std::to_string(proxy.port);
 
     if (proxy.scheme == "http" || proxy.scheme == "socks5" || proxy.scheme == "socks5h") {
@@ -366,6 +368,7 @@ export ProxyTunnel proxy_tunnel(const ProxyConfig& proxy,
         if (!tunnel.error.empty()) tunnel.socket.close();
     } else if (proxy.scheme == "https") {
         auto tls = std::make_unique<TlsSocket>();
+        tls->set_stop(stop);
         if (!tls->connect(proxy.host.c_str(), proxy.port, timeoutMs, verifySsl)) {
             // The session says why when the TCP connection was up, which is
             // where a proxy certificate that does not verify is refused.
@@ -385,11 +388,11 @@ export ProxyTunnel proxy_tunnel(const ProxyConfig& proxy,
 // say why it failed; `proxy_tunnel` can.
 export Socket proxy_connect(std::string_view proxyHost, int proxyPort,
                             std::string_view targetHost, int targetPort,
-                            int timeoutMs) {
+                            int timeoutMs, std::stop_token stop = {}) {
     ProxyConfig proxy;
     proxy.host = std::string(proxyHost);
     proxy.port = proxyPort;
-    auto tunnel = proxy_tunnel(proxy, targetHost, targetPort, timeoutMs);
+    auto tunnel = proxy_tunnel(proxy, targetHost, targetPort, timeoutMs, true, stop);
     return std::move(tunnel.socket);
 }