fix(http): time out stalled transactions without killing switched protocols

src/common/common.hpp

@@ -56,6 +56,23 @@ struct def_bool {
 	}
 };
 
+template <typename T, T default_value>
+struct def_value {
+	T value;
+
+	def_value() :
+		value(default_value) {
+	}
+
+	def_value(T init_value) :
+		value(init_value) {
+	}
+
+	operator T() const {
+		return value;
+	}
+};
+
 using StringPair = std::pair<string, string>;
 using ExpectedStringPair = expected::expected<StringPair, error::Error>;
 

src/common/http.hpp

@@ -396,6 +396,10 @@ class OutgoingResponse :
 template <typename StreamType>
 class BodyAsyncReader;
 
+// Long enough not to disturb slow but healthy transfers, short enough that a device recovers
+// on its own.
+static constexpr int kDefaultStreamTimeoutSeconds = 300;
+
 // Master object that connections are made from. Configure TLS options on this object before making
 // connections.
 struct ClientConfig {
@@ -412,6 +416,14 @@ struct ClientConfig {
 	string https_proxy;
 	string no_proxy;
 	string ssl_engine;
+
+	// Per-operation timeout on the underlying socket. Without it a peer which accepts a request
+	// and never answers wedges the client permanently. Does not apply to connections handed over
+	// after 101 Switching Protocols; see `Client::DisarmStreamTimeout()`.
+	//
+	// Uses def_value for the same reason as skip_verify above: a default member initializer would
+	// stop this being an aggregate under C++11, which Debug builds use on purpose.
+	common::def_value<int, kDefaultStreamTimeoutSeconds> stream_timeout_seconds;
 };
 
 enum class TransactionStatus {
@@ -567,6 +579,12 @@ class Client :
 	error::Error Initialize();
 	void DoCancel();
 
+	// Armed before each operation of a transaction. Must be disarmed before the socket is handed
+	// over after 101 Switching Protocols: Beast closes it on expiry regardless of who owns it by
+	// then, which would tear down a healthy long lived connection.
+	void ArmStreamTimeout();
+	void DisarmStreamTimeout();
+
 	void CallHandler(ResponseHandler handler);
 	void CallErrorHandler(
 		const error_code &ec, const OutgoingRequestPtr &req, ResponseHandler handler);

src/common/http/platform/beast/http.cpp

@@ -581,6 +581,11 @@ io::ExpectedAsyncReadWriterPtr Client::SwitchProtocol(IncomingResponsePtr req) {
 	*cancelled_ = true;
 	cancelled_ = make_shared<bool>(false);
 
+	// Required, and not covered by the 101 branch of `ReadHeaderHandler()`: callers may switch
+	// from the *header* handler, which runs before that branch and cancels it. The forwarder does
+	// exactly that, so this is what keeps the timeout off mender-connect's socket.
+	DisarmStreamTimeout();
+
 	auto stream = stream_;
 	// This no longer belongs to us.
 	stream_.reset();
@@ -626,6 +631,23 @@ void Client::CallErrorHandler(
 		err.WithContext(MethodToString(req->method_) + " " + req->orig_address_)));
 }
 
+// `next_layer().next_layer()` is the `beast::tcp_stream` inside
+// `ssl::stream<ssl::stream<beast::tcp_stream>>`, where the timeout lives in every socket mode.
+void Client::ArmStreamTimeout() {
+	if (!stream_) {
+		return;
+	}
+	stream_->next_layer().next_layer().expires_after(
+		chrono::seconds(client_config_.stream_timeout_seconds));
+}
+
+void Client::DisarmStreamTimeout() {
+	if (!stream_) {
+		return;
+	}
+	stream_->next_layer().next_layer().expires_never();
+}
+
 void Client::ResolveHandler(
 	const error_code &ec, const asio::ip::tcp::resolver::results_type &results) {
 	if (ec) {
@@ -679,6 +701,8 @@ void Client::ResolveHandler(
 
 	auto &cancelled = cancelled_;
 
+	// No timeout armed here: this connects on `lowest_layer()`, the raw socket, which bypasses the
+	// `beast::tcp_stream` timeout. Connect is bounded by the OS; later operations arm their own.
 	asio::async_connect(
 		stream_->lowest_layer(),
 		resolver_results_,
@@ -752,6 +776,8 @@ void Client::HandshakeHandler(
 
 	auto &cancelled = cancelled_;
 
+	ArmStreamTimeout();
+
 	stream.async_handshake(
 		ssl::stream_base::client, [this, cancelled, endpoint](const error_code &ec) {
 			if (*cancelled) {
@@ -810,6 +836,8 @@ void Client::ConnectHandler(const error_code &ec, const asio::ip::tcp::endpoint
 		}
 	};
 
+	ArmStreamTimeout();
+
 	switch (socket_mode_) {
 	case SocketMode::TlsTls:
 		http::async_write_header(*stream_, *request_data_.http_request_serializer_, handler);
@@ -947,6 +975,8 @@ void Client::WriteBody() {
 		}
 	};
 
+	ArmStreamTimeout();
+
 	switch (socket_mode_) {
 	case SocketMode::TlsTls:
 		http::async_write_some(*stream_, *request_data_.http_request_serializer_, handler);
@@ -972,6 +1002,9 @@ void Client::ReadHeader() {
 		}
 	};
 
+	// Without this, a server which never answers leaves this read outstanding forever.
+	ArmStreamTimeout();
+
 	switch (socket_mode_) {
 	case SocketMode::TlsTls:
 		http::async_read_some(
@@ -1052,6 +1085,10 @@ void Client::ReadHeaderHandler(const error_code &ec, size_t num_read) {
 				// Make an exception for 101 Switching Protocols response, where the TCP connection
 				// is meant to be reused.
 				DoCancel();
+			} else {
+				// About to be handed over (see `SwitchProtocol()`) and legitimately idle for long
+				// stretches, so the timeout must not follow it.
+				DisarmStreamTimeout();
 			}
 			CallHandler(body_handler_);
 		}
@@ -1163,10 +1200,7 @@ void Client::AsyncReadNextBodyPart(
 	auto &cancelled = cancelled_;
 	auto &response_data = response_data_;
 
-	// Set timeout to 5 minutes to ensure we don't hang during async read
-	// `next_layer().next_layer()` accesses the `beast::tcp_stream` from
-	// `ssl::stream<ssl::stream<beast::tcp_stream>>`
-	stream_->next_layer().next_layer().expires_after(chrono::minutes(5));
+	ArmStreamTimeout();
 
 	auto async_handler = [this, cancelled, response_data](const error_code &ec, size_t num_read) {
 		if (!*cancelled) {

tests/src/common/http_test.cpp

@@ -330,6 +330,71 @@ TEST(HttpTest, TestMissingResponse) {
 	EXPECT_EQ(http::TestInspector::GetStreams(server).size(), 1);
 }
 
+TEST(HttpTest, TestStalledResponseTimesOut) {
+	// A server which holds the connection open without answering, unlike one which drops it (see
+	// `TestMissingResponse`). Nothing else imposes a deadline, so the read used to stay
+	// outstanding forever.
+	TestEventLoop loop;
+
+	// Keeping the response alive is what stops the server tearing the connection down.
+	http::OutgoingResponsePtr stalled_resp;
+
+	http::ServerConfig server_config;
+	http::Server server(server_config, loop);
+	auto err = server.AsyncServeUrl(
+		"http://127.0.0.1:" TEST_PORT,
+		[](http::ExpectedIncomingRequestPtr exp_req) {
+			ASSERT_TRUE(exp_req) << exp_req.error().String();
+		},
+		[&stalled_resp](http::ExpectedIncomingRequestPtr exp_req) {
+			ASSERT_TRUE(exp_req) << exp_req.error().String();
+
+			auto exp_resp = exp_req.value()->MakeResponse();
+			ASSERT_TRUE(exp_resp) << exp_resp.error().String();
+
+			// Deliberately never reply.
+			stalled_resp = exp_resp.value();
+		});
+	ASSERT_EQ(err, error::NoError);
+
+	http::ClientConfig client_config;
+	client_config.stream_timeout_seconds = 1;
+	http::Client client(client_config, loop);
+
+	auto req = make_shared<http::OutgoingRequest>();
+	req->SetMethod(http::Method::GET);
+	req->SetAddress("http://127.0.0.1:" TEST_PORT);
+
+	bool header_handler_called = false;
+	auto start = chrono::steady_clock::now();
+	chrono::milliseconds elapsed {0};
+
+	err = client.AsyncCall(
+		req,
+		[&loop, &header_handler_called, &start, &elapsed](
+			http::ExpectedIncomingResponsePtr exp_resp) {
+			header_handler_called = true;
+			elapsed =
+				chrono::duration_cast<chrono::milliseconds>(chrono::steady_clock::now() - start);
+
+			EXPECT_FALSE(exp_resp) << "Expected the stalled request to fail, but got a response";
+
+			loop.Stop();
+		},
+		[](http::ExpectedIncomingResponsePtr exp_resp) {
+			FAIL() << "Should never receive a body";
+		});
+	ASSERT_EQ(err, error::NoError);
+
+	// Without the timeout the request never completes and `TestEventLoop` aborts the test.
+	loop.Run();
+
+	EXPECT_TRUE(header_handler_called);
+
+	// Guards against the connection having failed for some other reason instead.
+	EXPECT_GE(elapsed, chrono::milliseconds(900)) << "Failed too early to be the stream timeout";
+}
+
 TEST(HttpTest, TestDestroyResponseBeforeUsingIt) {
 	TestEventLoop loop;
 

tests/src/mender-auth/http_forwarder/http_forwarder_test.cpp

@@ -12,6 +12,8 @@
 //    See the License for the specific language governing permissions and
 //    limitations under the License.
 
+#include <algorithm>
+
 #include <gtest/gtest.h>
 #include <gmock/gmock.h>
 
@@ -642,6 +644,134 @@ TEST(HttpForwarderTests, ProtocolSwitch) {
 	EXPECT_EQ(copies, 2);
 }
 
+TEST(HttpForwarderTests, ProtocolSwitchSurvivesIdlePastStreamTimeout) {
+	// Regression test for MEN-9433: an earlier fix armed Beast's stream timeout and never disarmed
+	// it, so it followed the socket to its new owner and closed a healthy connection on expiry.
+	// This is how mender-connect's WebSocket is forwarded, so it is idle for long stretches.
+	const int timeout_seconds = 1;
+	const auto idle_time = chrono::seconds(timeout_seconds * 3);
+
+	// Generous, since the test deliberately spends most of its time waiting.
+	mtesting::TestEventLoop loop {chrono::seconds(30)};
+
+	const vector<uint8_t> payload {'i', 'd', 'l', 'e', '-', 'o', 'k'};
+
+	io::AsyncReadWriterPtr client_socket, server_socket;
+	events::Timer idle_timer {loop};
+	vector<uint8_t> received(payload.size());
+	size_t bytes_read = 0;
+	bool wrote_after_idle = false;
+	bool read_after_idle = false;
+	// Recorded rather than asserted in the handlers, so a failure still stops the loop.
+	error::Error write_error = error::NoError;
+	error::Error read_error = error::NoError;
+
+	http::ServerConfig server_config;
+	http::Server server(server_config, loop);
+
+	auto err = server.AsyncServeUrl(
+		"http://127.0.0.1:" TEST_PORT,
+		[](http::ExpectedIncomingRequestPtr exp_req) {
+			ASSERT_TRUE(exp_req) << exp_req.error().String();
+		},
+		[&](http::ExpectedIncomingRequestPtr exp_req) {
+			ASSERT_TRUE(exp_req) << exp_req.error().String();
+
+			auto exp_resp = exp_req.value()->MakeResponse();
+			ASSERT_TRUE(exp_resp) << exp_resp.error().String();
+			auto resp = exp_resp.value();
+
+			resp->SetStatusCodeAndMessage(101, "Switching Protocols");
+			auto err = resp->AsyncSwitchProtocol([&](io::ExpectedAsyncReadWriterPtr exp_socket) {
+				ASSERT_TRUE(exp_socket) << exp_socket.error().String();
+				server_socket = exp_socket.value();
+
+				// Sit idle well past the timeout, then write. A timeout left armed anywhere along
+				// the forwarded path closes the socket before this runs.
+				idle_timer.AsyncWait(idle_time, [&](error::Error err) {
+					ASSERT_EQ(err, error::NoError);
+
+					auto write_err = server_socket->AsyncWrite(
+						payload.begin(), payload.end(), [&](io::ExpectedSize result) {
+							if (!result) {
+								write_error = result.error();
+								// The read will never complete now, so end the test here.
+								loop.Stop();
+								return;
+							}
+							wrote_after_idle = true;
+						});
+					EXPECT_EQ(write_err, error::NoError);
+				});
+			});
+			ASSERT_EQ(err, error::NoError);
+		});
+	ASSERT_EQ(err, error::NoError);
+
+	http::ClientConfig client_config;
+	client_config.stream_timeout_seconds = timeout_seconds;
+
+	hf::TestServer forwarder(server_config, client_config, loop);
+	err = forwarder.AsyncForward("http://127.0.0.1:0", "http://127.0.0.1:" TEST_PORT "/");
+	ASSERT_EQ(err, error::NoError);
+
+	http::Client client(client_config, loop);
+	auto req = make_shared<http::OutgoingRequest>();
+	req->SetMethod(http::Method::GET);
+	req->SetAddress(http::JoinUrl(forwarder.GetUrl(), "/test-endpoint"));
+
+	err = client.AsyncCall(
+		req,
+		[](http::ExpectedIncomingResponsePtr exp_resp) {
+			ASSERT_TRUE(exp_resp) << exp_resp.error().String();
+		},
+		[&](http::ExpectedIncomingResponsePtr exp_resp) {
+			ASSERT_TRUE(exp_resp) << exp_resp.error().String();
+			ASSERT_EQ(exp_resp.value()->GetStatusCode(), 101);
+
+			auto exp_socket = exp_resp.value()->SwitchProtocol();
+			ASSERT_TRUE(exp_socket) << exp_socket.error().String();
+			client_socket = exp_socket.value();
+
+			// Post the read straight away and let it sit; it has to survive the idle period.
+			auto read_err = client_socket->AsyncRead(
+				received.begin(), received.end(), [&](io::ExpectedSize result) {
+					if (!result) {
+						read_error = result.error();
+					} else {
+						bytes_read = result.value();
+						read_after_idle = true;
+					}
+					loop.Stop();
+				});
+			EXPECT_EQ(read_err, error::NoError);
+		});
+	ASSERT_EQ(err, error::NoError);
+
+	loop.Run();
+
+	// A stale timeout closes the socket, surfacing as an error on whichever side touches it first.
+	// These are the assertions that matter.
+	EXPECT_EQ(write_error, error::NoError) << "socket closed by a stale timeout?";
+	EXPECT_EQ(read_error, error::NoError) << "socket closed by a stale timeout?";
+	EXPECT_TRUE(wrote_after_idle);
+	EXPECT_TRUE(read_after_idle);
+
+	// The byte count is incidental (a short read is legal), so only compare what arrived.
+	EXPECT_GT(bytes_read, 0U);
+	ASSERT_LE(bytes_read, payload.size());
+	EXPECT_TRUE(std::equal(
+		received.begin(), received.begin() + static_cast<ptrdiff_t>(bytes_read), payload.begin()))
+		<< "Data received after the idle period does not match what was sent";
+
+	if (server_socket) {
+		server_socket->Cancel();
+	}
+	if (client_socket) {
+		client_socket->Cancel();
+	}
+}
+
 TEST(HttpForwarderTests, SocketMemoryLeaks) {
 	// Intentionally return while the sockets are still open to make sure no memory is leaked
 	// (relying on Address Sanitizer here).