bugfixes (e1d0a26f) · Commits · jan.koester / libhttppp

src/http.cpp

+126 −35

Changes for src/http.cpp: 126 added lines, 35 removed lines.
Original line number	Diff line number	Diff line
		@@ -40,6 +40,7 @@
		#include <set>
		#include <mutex>
		#include <iomanip>
		#include <optional>
		#include <random>

		#include "config.h"
		@@ -338,6 +339,8 @@ void libhttppp::HttpClient::resetConnection(){
		// Reset HTTP/2 connection state on any new connection
		_h2PrefaceSent = false;
		_h2NextStreamId = 1;
		_h2PeerConnWindow = 65535;
		_h2PeerInitialStreamWindow = 65535;
		_h2Decoder.reset();
		_peerWantsClose = false;
		// getProtocol() alone isn't enough: it's pinned to HTTP3 for the object's
		@@ -846,6 +849,7 @@ static void h3AppendFrame(std::vector<uint8_t> &out, uint64_t type, const std::v

		static constexpr uint8_t H2C_FRAME_DATA = 0x00;
		static constexpr uint8_t H2C_FRAME_HEADERS = 0x01;
		static constexpr uint8_t H2C_FRAME_RST_STREAM = 0x03;
		static constexpr uint8_t H2C_FRAME_SETTINGS = 0x04;
		static constexpr uint8_t H2C_FRAME_PING = 0x06;
		static constexpr uint8_t H2C_FRAME_GOAWAY = 0x07;
		@@ -2604,13 +2608,23 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		raw.reserve(16384);

		if (hasBufferedBody \|\| bodySource) {
		// Track the peer's (server's) flow control window.
		// Initial value is 65535 until we receive the server's SETTINGS.
		int32_t peer_conn_window = 65535;
		int32_t peer_stream_window = 65535;

		// Helper: parse buffered frames for SETTINGS / WINDOW_UPDATE / PING
		// The server's flow control windows: the connection's carries over from earlier
		// requests on this connection, a new stream starts at the server's
		// SETTINGS_INITIAL_WINDOW_SIZE (65535 until its SETTINGS arrive).
		int64_t &peer_conn_window = _h2PeerConnWindow;
		int64_t peer_stream_window = _h2PeerInitialStreamWindow;

		// Set when the server is done with our request body before we are: it reset the
		// stream (RST_STREAM, e.g. NO_ERROR after an early answer, RFC 9113 §8.1) or
		// already sent its complete response. The send loops stop and go read it.
		bool peer_stopped = false;
		bool peer_reset = false;

		// Helper: handle buffered SETTINGS / WINDOW_UPDATE / PING frames; every other
		// frame (the response's HEADERS/DATA, RST_STREAM) stays queued, in order, for
		// the response phase -- an early answer must not be thrown away here.
		auto process_fc_frames = [&]() {
		std::vector<uint8_t> kept;
		size_t foff = 0;
		while (foff + H2C_FRAME_HEADER_LEN <= raw.size()) {
		uint32_t flen = (static_cast<uint32_t>(raw[foff]) << 16)
		@@ -2634,9 +2648,10 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		\| (static_cast<uint32_t>(fp[si+4]) << 8)
		\| fp[si+5];
		if (setting_id == 0x04) { // SETTINGS_INITIAL_WINDOW_SIZE
		int32_t delta = static_cast<int32_t>(setting_val)
		- peer_stream_window;
		peer_stream_window += delta;
		// RFC 9113 §6.9.2: adjusts open streams by the difference.
		peer_stream_window += static_cast<int64_t>(setting_val)
		- _h2PeerInitialStreamWindow;
		_h2PeerInitialStreamWindow = setting_val;
		}
		}
		send_all(h2cBuildSettingsAck());
		@@ -2647,21 +2662,33 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		\| fp[3];
		if (inc > 0) {
		if (fsid == 0)
		peer_conn_window += static_cast<int32_t>(inc);
		peer_conn_window += inc;
		else if (fsid == stream_id)
		peer_stream_window += static_cast<int32_t>(inc);
		peer_stream_window += inc;
		}
		} else if (ftype == H2C_FRAME_PING && !(fflags & H2C_FLAG_ACK)
		&& flen == 8) {
		std::string pp(reinterpret_cast<const char*>(fp), flen);
		send_all(h2cBuildFrame(H2C_FRAME_PING, H2C_FLAG_ACK, 0, pp));
		} else if (ftype != H2C_FRAME_SETTINGS && ftype != H2C_FRAME_WINDOW_UPDATE
		&& ftype != H2C_FRAME_PING) {
		if (fsid == stream_id) {
		if (ftype == H2C_FRAME_RST_STREAM) {
		peer_stopped = peer_reset = true;
		} else if ((ftype == H2C_FRAME_HEADERS \|\| ftype == H2C_FRAME_DATA)
		&& (fflags & H2C_FLAG_END_STREAM)) {
		peer_stopped = true;
		}
		}
		kept.insert(kept.end(), raw.begin() + static_cast<ptrdiff_t>(foff),
		raw.begin() + static_cast<ptrdiff_t>(foff + H2C_FRAME_HEADER_LEN + flen));
		}
		// Leave HEADERS / DATA in the buffer for the response phase
		// but only consume control frames here.
		foff += H2C_FRAME_HEADER_LEN + flen;
		}
		if (foff > 0)
		raw.erase(raw.begin(), raw.begin() + static_cast<ptrdiff_t>(foff));
		if (foff > 0) {
		kept.insert(kept.end(), raw.begin() + static_cast<ptrdiff_t>(foff), raw.end());
		raw.swap(kept);
		}
		};

		// Try a non-blocking read to pick up the server's SETTINGS
		@@ -2679,10 +2706,10 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		}

		// Blocks until the peer has granted some window, returns how much.
		auto wait_window = [&]() -> int32_t {
		int32_t allowed = std::min(peer_conn_window, peer_stream_window);
		auto wait_window = [&]() -> int64_t {
		int64_t allowed = std::min(peer_conn_window, peer_stream_window);
		// If window exhausted, block-read for WINDOW_UPDATE
		while (allowed <= 0) {
		while (allowed <= 0 && !peer_stopped) {
		size_t n = recv_blocking(_recvScratch);
		if (n == 0) {
		HTTPException he;
		@@ -2700,7 +2727,7 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(

		if (bodySource) {
		std::vector<char> chunkBuf(H2C_MAX_FRAME_SIZE);
		for (;;) {
		while (!peer_stopped) {
		size_t have = (*bodySource)(chunkBuf.data(), chunkBuf.size());
		if (have == 0) {
		// The source can only say a chunk was its last by returning 0
		@@ -2712,19 +2739,21 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		}
		size_t hoff = 0;
		while (hoff < have) {
		size_t chunk = std::min(have - hoff,
		static_cast<size_t>(wait_window()));
		int64_t allowed = wait_window();
		if (peer_stopped) break;
		size_t chunk = std::min(have - hoff, static_cast<size_t>(allowed));
		send_all(h2cBuildFrame(H2C_FRAME_DATA, 0, stream_id,
		std::string(chunkBuf.data() + hoff, chunk)));
		peer_conn_window -= static_cast<int32_t>(chunk);
		peer_stream_window -= static_cast<int32_t>(chunk);
		peer_conn_window -= static_cast<int64_t>(chunk);
		peer_stream_window -= static_cast<int64_t>(chunk);
		hoff += chunk;
		}
		}
		} else {
		size_t offset = 0;
		while (offset < postBody->size()) {
		int32_t allowed = wait_window();
		int64_t allowed = wait_window();
		if (peer_stopped) break;
		size_t chunk = std::min({postBody->size() - offset,
		static_cast<size_t>(allowed),
		H2C_MAX_FRAME_SIZE});
		@@ -2732,11 +2761,17 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		uint8_t flags = last ? H2C_FLAG_END_STREAM : 0;
		std::string payload(postBody->data() + offset, chunk);
		send_all(h2cBuildFrame(H2C_FRAME_DATA, flags, stream_id, payload));
		peer_conn_window -= static_cast<int32_t>(chunk);
		peer_stream_window -= static_cast<int32_t>(chunk);
		peer_conn_window -= static_cast<int64_t>(chunk);
		peer_stream_window -= static_cast<int64_t>(chunk);
		offset += chunk;
		}
		}
		if (peer_stopped && !peer_reset) {
		// Our half is still open with the body unfinished: close it.
		uint8_t rst[4] = {0, 0, 0, 0x08}; // CANCEL
		send_all(h2cBuildFrame(H2C_FRAME_RST_STREAM, 0, stream_id,
		std::string(reinterpret_cast<char*>(rst), 4)));
		}
		}

		// 4) Read response frames
		@@ -2822,7 +2857,16 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		switch (frame_type) {
		case H2C_FRAME_SETTINGS: {
		if (!(frame_flags & H2C_FLAG_ACK)) {
		// Server SETTINGS — acknowledge it
		// Server SETTINGS — remember the stream window for later requests,
		// then acknowledge it.
		for (size_t si = 0; si + 6 <= frame_len; si += 6) {
		uint16_t id = (static_cast<uint16_t>(payload[si]) << 8) \| payload[si+1];
		if (id == 0x04)
		_h2PeerInitialStreamWindow =
		(static_cast<uint32_t>(payload[si+2]) << 24) \|
		(static_cast<uint32_t>(payload[si+3]) << 16) \|
		(static_cast<uint32_t>(payload[si+4]) << 8) \| payload[si+5];
		}
		send_all(h2cBuildSettingsAck());
		}
		break;
		@@ -2871,6 +2915,13 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		}

		case H2C_FRAME_WINDOW_UPDATE:
		// Connection credit keeps counting for the next request's body.
		if (frame_stream == 0 && frame_len >= 4) {
		_h2PeerConnWindow += ((static_cast<uint32_t>(payload[0]) & 0x7f) << 24) \|
		(static_cast<uint32_t>(payload[1]) << 16) \|
		(static_cast<uint32_t>(payload[2]) << 8) \| payload[3];
		}
		break;
		case H2C_FRAME_PING: {
		if (frame_type == H2C_FRAME_PING && !(frame_flags & H2C_FLAG_ACK)) {
		// Respond to PING
		@@ -2880,6 +2931,23 @@ const std::vector<char> libhttppp::HttpClient::_h2Request(
		break;
		}

		case H2C_FRAME_RST_STREAM: {
		// Reaching here means our response isn't complete (see the loop
		// condition) -- the server aborted the stream, nothing more will come.
		if (frame_stream == stream_id) {
		uint32_t code = frame_len >= 4
		? (static_cast<uint32_t>(payload[0]) << 24) \|
		(static_cast<uint32_t>(payload[1]) << 16) \|
		(static_cast<uint32_t>(payload[2]) << 8) \| payload[3]
		: 0;
		HTTPException he;
		he[HTTPException::Error] << "HTTP/2: server reset the stream (error "
		<< code << ") before completing its response";
		throw he;
		}
		break;
		}

		case H2C_FRAME_GOAWAY: {
		// RFC 7540 §6.8: 4-byte Last-Stream-ID (top bit reserved)
		// followed by a 4-byte error code. Reaching this case means
		@@ -3413,6 +3481,22 @@ const std::vector<char> libhttppp::HttpClient::RequestStream(int requestType,

		std::vector<char> buf(64 * 1024);
		uint64_t sent = 0;
		// A server may answer before taking the whole body and then close (RFC 9112 §9.6):
		// if sending fails, an answer may already be waiting -- it's the better result.
		std::optional<std::vector<char>> early;
		auto send = [&](const char *p, size_t len) -> bool {
		try {
		_sendAll(p, len);
		return true;
		} catch (const std::exception &) {
		try {
		early = _h1ReadResponse(method);
		} catch (...) {
		}
		if (!early) throw; // nothing readable: report the send failure
		return false;
		}
		};
		for (;;) {
		size_t n = source(buf.data(), buf.size());
		if (n == 0)
		@@ -3427,16 +3511,23 @@ const std::vector<char> libhttppp::HttpClient::RequestStream(int requestType,
		if (chunked) {
		char sizeLine[32];
		int sl = std::snprintf(sizeLine, sizeof(sizeLine), "%zx\r\n", n);
		_sendAll(sizeLine, static_cast<size_t>(sl));
		_sendAll(buf.data(), n);
		_sendAll("\r\n", 2);
		} else {
		_sendAll(buf.data(), n);
		if (!send(sizeLine, static_cast<size_t>(sl)) \|\| !send(buf.data(), n) \|\|
		!send("\r\n", 2))
		break;
		} else if (!send(buf.data(), n)) {
		break;
		}
		}
		if (chunked) {
		_sendAll("0\r\n\r\n", 5);
		} else if (sent != static_cast<uint64_t>(contentLength)) {
		if (early) {
		// The connection is finished either way.
		resetConnection();
		return *early;
		}
		if (chunked && !send("0\r\n\r\n", 5)) {
		resetConnection();
		return *early;
		}
		if (!chunked && sent != static_cast<uint64_t>(contentLength)) {
		HTTPException ee;
		ee[HTTPException::Error] << "RequestStream: body ended after " << sent
		<< " of " << contentLength << " bytes";

src/http.h

+6 −0

Changes for src/http.h: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -275,6 +275,12 @@ namespace libhttppp {

		// HTTP/3 client helpers
		uint32_t _h2NextStreamId = 1; // next client-initiated stream ID (odd)
		// The server's flow-control state for this connection (RFC 9113 §6.9), kept across
		// requests: the connection send window and its SETTINGS_INITIAL_WINDOW_SIZE. Restarting
		// both at 65535 per request (as before) overran a server that hadn't replenished the
		// connection window yet, and ignored any credit/SETTINGS seen while reading a response.
		int64_t _h2PeerConnWindow = 65535;
		int64_t _h2PeerInitialStreamWindow = 65535;
		std::unique_ptr<hpack::Decoder> _h2Decoder; // persistent HPACK decoder for connection
		// bodySource (RequestStream()) replaces postBody when set -- the body is pulled and
		// framed incrementally instead of sent from one buffer.

src/httpd.cpp

+46 −7

Changes for src/httpd.cpp: 46 added lines, 7 removed lines.
Original line number	Diff line number	Diff line
		@@ -582,6 +582,30 @@ public:
		bool finished() const override { return _done; }
		void abort() override { if (!_done) _aborted = true; }

		// Caps every later read() at this point in time (lingering close, see drainUnreadBody).
		void setDeadline(std::chrono::steady_clock::time_point d) { _deadline = d; }

		// Reads and discards whatever is left of the body until it ends or `budget` runs out --
		// RFC 9112 §9.6 lingering close: after answering early, keep accepting the client's body
		// so it finishes sending and reads the answer, instead of hitting a broken pipe (or a
		// TCP RST that discards the answer before it's read). True = body fully consumed, the
		// connection can be reused.
		bool drainUnreadBody(std::chrono::steady_clock::duration budget) {
		setDeadline(std::chrono::steady_clock::now() + budget);
		// One blocking socket read mustn't outlast the budget either.
		if (auto &sock = _req.slots[0].csock)
		sock->setTimeout(static_cast<int>(
		std::chrono::duration_cast<std::chrono::milliseconds>(budget).count()));
		char scratch[65536];
		try {
		while (read(scratch, sizeof(scratch)) > 0) {
		}
		} catch (const std::exception &) {
		return false;
		}
		return _done;
		}

		private:
		enum class State { Size, Data, DataEnd, Trailer };

		@@ -607,7 +631,10 @@ private:
		}
		}
		netplus::buffer b(64 * 1024);
		const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(60);
		const auto deadline = std::min(_deadline,
		std::chrono::steady_clock::now() + std::chrono::seconds(60));
		if (std::chrono::steady_clock::now() >= deadline)
		fail("request body: timed out waiting for data");
		size_t n = 0;
		for (;;) {
		b.size = 64 * 1024; // recvData() shrinks it to what arrived
		@@ -673,6 +700,7 @@ private:
		bool _sendContinue = false;
		bool _done = false;
		bool _aborted = false;
		std::chrono::steady_clock::time_point _deadline = std::chrono::steady_clock::time_point::max();
		netplus::socketwait _sw;
		};

		@@ -965,11 +993,8 @@ void libhttppp::HttpEvent::_dispatchH1Request(HttpRequest &cureq, size_t consume
		// A streamed body the handler didn't read to the end leaves the rest on
		// the wire, where the next request's parse would pick it up as garbage
		// -- answer, then close instead of reattaching.
		bool closeAfterResponse = false;
		if (cureq2._bodyStream) {
		closeAfterResponse = !cureq2._bodyStream->finished();
		cureq2._bodyStream.reset();
		}
		std::shared_ptr<RequestBodyStream> bodyStream =
		std::move(cureq2._bodyStream);
		if (!cureq2.slots[0].csock) {
		// RequestEvent itself took over this connection's socket for
		// something outside the normal request/response cycle -- e.g. a
		@@ -1017,10 +1042,14 @@ void libhttppp::HttpEvent::_dispatchH1Request(HttpRequest &cureq, size_t consume
		return;
		}
		}
		if (closeAfterResponse) {
		if (bodyStream && !bodyStream->finished()) {
		// Answered without reading the whole body: lingering close.
		auto h1 = dynamic_cast<H1RequestBodyReader >(bodyStream.get());
		if (!h1 \|\| !h1->drainUnreadBody(std::chrono::seconds(10))) {
		cureq2.slots[0].csock->close();
		return;
		}
		}
		} catch (const std::exception &) {
		// Either RequestEvent or the flush hit a hard error (peer reset,
		// socket-level failure) -- nothing to reattach. csock can't have gone
		@@ -2917,6 +2946,16 @@ void libhttppp::HttpEvent::sendH3StreamResponse(netplus::socket *sock, uint64_t
		response.insert(response.end(), body.begin(), body.end());
		}
		q->sendStreamData(streamId, response, true);

		// onH3StreamHeaders/onH3DataChunk streams are answered through here: that's the
		// response side done, so the stream retires once its request side has ended too.
		std::shared_ptr<H3PullState> st;
		{
		std::lock_guard<std::mutex> lk(_h3BufferMutex);
		auto it = _h3PullStates.find({q, streamId});
		if (it != _h3PullStates.end()) st = it->second;
		}
		if (st) _h3StreamSideDone(q, streamId, st, false);
		}

		void libhttppp::HttpEvent::RequestEvent(netplus::con &curcon,const int tid,ULONG_PTR args){

test/httpd_request_body_stream_test.cpp

+65 −0

Changes for test/httpd_request_body_stream_test.cpp: 65 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -157,6 +157,9 @@ public:
		}
		res.setHeaderData("x-sha256")->push_back(sha256Hex(body));
		res.setHeaderData("x-streamed")->push_back(streamed ? "1" : "0");
		// The dispatcher only stashes a peer address on H2/H3 per-stream requests (an H1
		// request is the connection itself and has its socket).
		res.setHeaderData("x-h1")->push_back(req.getPeerAddress().empty() ? "1" : "0");
		std::string n = std::to_string(body.size());
		res.send(req, n, static_cast<int>(n.size()));
		}
		@@ -396,6 +399,25 @@ void h2Upload(H2Uploader &client, const std::string &path, const std::vector<cha
		}
		}

		// "/early" with a body far larger than any window or socket buffer: the server answers 413
		// without reading it. H1 relies on lingering close (the server drains the rest), H2 on
		// RST_STREAM(NO_ERROR) stopping the client's upload, H3 on STOP_SENDING. Then the same client
		// must still work.
		void earlyThenReuse(libhttppp::HttpClient &client, const std::string &label) {
		try {
		std::vector<char> data = randomBuffer(20u * 1024 * 1024);
		libhttppp::HttpRequest req;
		req.setRequestURL("/early");
		client.RequestStream(POSTREQUEST, req, sourceOf(data, 50000),
		static_cast<int64_t>(data.size()));
		check(client.lastStatusCode() == 413, label + ": early 413 received");
		} catch (std::exception &e) {
		check(false, label + ": early answer threw " + e.what());
		}
		streamedUpload(client, "/sum", randomBuffer(1048576), true, true,
		label + ": same client afterwards");
		}

		std::string recvSome(netplus::tcp &sock) {
		netplus::buffer b(4096);
		size_t n = sock.recvData(b, 0);
		@@ -445,6 +467,8 @@ int main() {
		}
		streamedUpload(client, "/buffered", randomBuffer(100000), true, false,
		"H1 buffered fallback");
		check(headerValue(client.lastResponse(), "x-h1") == "1", "H1 client: really HTTP/1.1");
		earlyThenReuse(client, "H1 early answer 20MB");
		}
		{
		// Expect: 100-continue -- the client holds the body back until the interim 100.
		@@ -479,6 +503,33 @@ int main() {
		}
		streamedUpload(client, "/buffered", randomBuffer(100000), true, false,
		"TLS buffered fallback");
		earlyThenReuse(client, "TLS H1 early answer 20MB");
		}

		// ---------------- HTTP/2 through HttpClient (TLS + ALPN) ----------------
		std::cout << "-- HTTP/2 via HttpClient (TLS) --" << std::endl;
		{
		libhttppp::HttpUrl url("https://127.0.0.1:" + std::to_string(kTlsPort) + "/");
		libhttppp::HttpClient client(url, /vers=/2, /timeoutSec=/20);
		streamedUpload(client, "/sum", randomBuffer(1), true, true, "H2 client size=1");
		check(headerValue(client.lastResponse(), "x-h1") == "0",
		"H2 client: ALPN negotiated h2 (not HTTP/1.1)");
		for (size_t size : {size_t(65536), size_t(1048576), kBig}) {
		std::vector<char> data = randomBuffer(size);
		std::string s = std::to_string(size);
		streamedUpload(client, "/sum", data, true, true, "H2 client known length size=" + s);
		streamedUpload(client, "/sum", data, false, true, "H2 client unknown length size=" + s);
		}
		// More than one connection window's worth on one connection: the client's
		// connection-window bookkeeping must carry over between requests.
		for (int i = 0; i < 3; ++i)
		streamedUpload(client, "/sum", randomBuffer(8u * 1024 * 1024), true, true,
		"H2 client reuse #" + std::to_string(i));
		earlyThenReuse(client, "H2 client early answer 20MB");
		streamedUpload(client, "/buffered", randomBuffer(100000), true, false,
		"H2 client buffered");
		check(headerValue(client.lastResponse(), "x-h1") == "0",
		"H2 client: still h2 at the end");
		}

		// ---------------- HTTP/3 ----------------
		@@ -523,6 +574,20 @@ int main() {
		} catch (std::exception &e) {
		check(false, std::string("H3 onH3DataChunk: threw ") + e.what());
		}
		earlyThenReuse(client, "H3 early answer 20MB");

		// Every request stream above -- streamed, buffered, callback-mode, early-answered
		// -- must have been retired on the server by now (it used to keep all of them, with
		// their buffers, for the connection's lifetime). Retirement of the early-answered
		// one waits on the client's RESET_STREAM, so allow a moment.
		size_t open = 0;
		for (int i = 0; i < 50; ++i) {
		open = quicSock->childOpenStreamCount(/bidirectionalOnly=/true);
		if (open == 0) break;
		std::this_thread::sleep_for(std::chrono::milliseconds(100));
		}
		check(open == 0, "H3 server retired every request stream (open: " +
		std::to_string(open) + ")");
		}

		// ---------------- HTTP/2 (h2c) ----------------