Add unit-tests for PerfdataWriterConnection

There's a set of two tests for each perfdatawriter, just
to make sure they can connect and send data that looks reasonably
correct, and to make sure pausing actually works while the connection
is stuck.

Then there's a more in-depth suite of tests for PerfdataWriterConnection
itself, to verify that connection handling works well in all types
of scenarios.

Co-authored-by: Yonas Habteab <yonas.habteab@icinga.com>

test/CMakeLists.txt

@@ -140,6 +140,18 @@ if(ICINGA2_WITH_NOTIFICATION)
   )
 endif()
 
+if(ICINGA2_WITH_PERFDATA)
+  list(APPEND base_test_SOURCES
+    perfdata-elasticsearchwriter.cpp
+    perfdata-gelfwriter.cpp
+    perfdata-graphitewriter.cpp
+    perfdata-influxdbwriter.cpp
+    perfdata-opentsdbwriter.cpp
+    perfdata-perfdatawriterconnection.cpp
+    $<TARGET_OBJECTS:perfdata>
+  )
+endif()
+
 if(ICINGA2_UNITY_BUILD)
   mkunity_target(base test base_test_SOURCES)
 endif()

test/perfdata-elasticsearchwriter.cpp

@@ -0,0 +1,57 @@
+// SPDX-FileCopyrightText: 2026 Icinga GmbH <https://icinga.com>
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include <BoostTestTargetConfig.h>
+#include "perfdata/elasticsearchwriter.hpp"
+#include "test/base-testloggerfixture.hpp"
+#include "test/perfdata-perfdatawriterfixture.hpp"
+#include "test/utils.hpp"
+
+using namespace icinga;
+
+BOOST_FIXTURE_TEST_SUITE(perfdata_elasticsearchwriter, PerfdataWriterFixture<ElasticsearchWriter>,
+	*boost::unit_test::label("perfdata")
+	*boost::unit_test::label("network")
+)
+
+BOOST_AUTO_TEST_CASE(connect)
+{
+	ResumeWriter();
+
+	ReceiveCheckResults(1, ServiceState::ServiceCritical);
+
+	Accept();
+	auto resp = GetSplitDecodedRequestBody();
+	SendResponse();
+
+	// ElasticsearchWriter wants to send the same message twice, once for the check result
+	// and once for the "state change".
+	resp = GetSplitDecodedRequestBody();
+	SendResponse();
+
+	// Just some basic sanity tests. It's not important to check if everything is entirely
+	// correct here.
+	BOOST_REQUIRE_GT(resp->GetLength(), 1);
+	Dictionary::Ptr cr = resp->Get(1);
+	BOOST_CHECK(cr->Contains("@timestamp"));
+	BOOST_CHECK_EQUAL(cr->Get("check_command"), "dummy");
+	BOOST_CHECK_EQUAL(cr->Get("host"), "h1");
+
+	PauseWriter();
+}
+
+BOOST_AUTO_TEST_CASE(pause_with_pending_work)
+{
+	ResumeWriter();
+
+	// Process check-results until the writer is stuck.
+	BOOST_REQUIRE_MESSAGE(GetWriterStuck(10s), "Failed to get Writer stuck.");
+
+	// Now try to pause.
+	PauseWriter();
+
+	REQUIRE_LOG_MESSAGE("Connection stopped\\.", 10s);
+	REQUIRE_LOG_MESSAGE("'ElasticsearchWriter' paused\\.", 10s);
+}
+
+BOOST_AUTO_TEST_SUITE_END()

test/perfdata-gelfwriter.cpp

@@ -0,0 +1,48 @@
+// SPDX-FileCopyrightText: 2026 Icinga GmbH <https://icinga.com>
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include <BoostTestTargetConfig.h>
+#include "perfdata/gelfwriter.hpp"
+#include "test/base-testloggerfixture.hpp"
+#include "test/perfdata-perfdatawriterfixture.hpp"
+#include "test/utils.hpp"
+
+using namespace icinga;
+
+BOOST_FIXTURE_TEST_SUITE(perfdata_gelfwriter, PerfdataWriterFixture<GelfWriter>,
+	*boost::unit_test::label("perfdata")
+	*boost::unit_test::label("network")
+)
+
+BOOST_AUTO_TEST_CASE(connect)
+{
+	ResumeWriter();
+
+	ReceiveCheckResults(1, ServiceState::ServiceCritical);
+
+	Accept();
+	Dictionary::Ptr resp = JsonDecode(GetDataUntil('\0'));
+
+	// Just some basic sanity tests. It's not important to check if everything is entirely
+	// correct here.
+	BOOST_CHECK_CLOSE(resp->Get("timestamp").Get<double>(), Utility::GetTime(), 0.5);
+	BOOST_CHECK_EQUAL(resp->Get("_check_command"), "dummy");
+	BOOST_CHECK_EQUAL(resp->Get("_hostname"), "h1");
+	PauseWriter();
+}
+
+BOOST_AUTO_TEST_CASE(pause_with_pending_work)
+{
+	ResumeWriter();
+
+	// Process check-results until the writer is stuck.
+	BOOST_REQUIRE_MESSAGE(GetWriterStuck(10s), "Failed to get Writer stuck.");
+
+	// Now stop reading and try to pause OpenTsdbWriter.
+	PauseWriter();
+
+	REQUIRE_LOG_MESSAGE("Connection stopped\\.", 1s);
+	REQUIRE_LOG_MESSAGE("'GelfWriter' paused\\.", 1s);
+}
+
+BOOST_AUTO_TEST_SUITE_END()

test/perfdata-graphitewriter.cpp

@@ -0,0 +1,47 @@
+// SPDX-FileCopyrightText: 2026 Icinga GmbH <https://icinga.com>
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include <BoostTestTargetConfig.h>
+#include "base/perfdatavalue.hpp"
+#include "perfdata/graphitewriter.hpp"
+#include "test/base-testloggerfixture.hpp"
+#include "test/perfdata-perfdatawriterfixture.hpp"
+#include "test/utils.hpp"
+
+using namespace icinga;
+
+BOOST_FIXTURE_TEST_SUITE(perfdata_graphitewriter, PerfdataWriterFixture<GraphiteWriter>,
+	*boost::unit_test::label("perfdata")
+	*boost::unit_test::label("network")
+)
+
+BOOST_AUTO_TEST_CASE(connect)
+{
+	ResumeWriter();
+
+	ReceiveCheckResults(1, ServiceState::ServiceCritical);
+
+	Accept();
+	auto msg = GetDataUntil('\n');
+
+	// Just some basic sanity tests. It's not important to check if everything is entirely correct here.
+	std::string_view cmpStr{"icinga2.h1.host.dummy.perfdata.dummy.value 42"};
+	BOOST_REQUIRE_EQUAL(msg.substr(0, cmpStr.length()), cmpStr);
+	PauseWriter();
+}
+
+BOOST_AUTO_TEST_CASE(pause_with_pending_work)
+{
+	ResumeWriter();
+
+	// Process check-results until the writer is stuck.
+	BOOST_REQUIRE_MESSAGE(GetWriterStuck(10s), "Failed to get Writer stuck.");
+
+	// Now stop reading and try to pause OpenTsdbWriter.
+	PauseWriter();
+
+	REQUIRE_LOG_MESSAGE("Connection stopped\\.", 10s);
+	REQUIRE_LOG_MESSAGE("'GraphiteWriter' paused\\.", 10s);
+}
+
+BOOST_AUTO_TEST_SUITE_END()

test/perfdata-influxdbwriter.cpp

@@ -0,0 +1,50 @@
+// SPDX-FileCopyrightText: 2026 Icinga GmbH <https://icinga.com>
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include <BoostTestTargetConfig.h>
+#include "perfdata/influxdb2writer.hpp"
+#include "test/base-testloggerfixture.hpp"
+#include "test/perfdata-perfdatawriterfixture.hpp"
+
+using namespace icinga;
+
+BOOST_FIXTURE_TEST_SUITE(perfdata_influxdbwriter, PerfdataWriterFixture<Influxdb2Writer>,
+	*boost::unit_test::label("perfdata")
+	*boost::unit_test::label("network")
+)
+
+BOOST_AUTO_TEST_CASE(connect)
+{
+	ResumeWriter();
+
+	ReceiveCheckResults(1, ServiceState::ServiceCritical);
+
+	Accept();
+	auto req = GetSplitRequestBody(',');
+	SendResponse(boost::beast::http::status::no_content);
+
+	// Just some basic sanity tests. It's not important to check if everything is entirely
+	// correct here.
+	BOOST_REQUIRE_EQUAL(req.size(), 3);
+	BOOST_CHECK_EQUAL(req[0], "dummy");
+	BOOST_CHECK_EQUAL(req[1], "hostname=h1");
+	std::string_view perfData = "metric=dummy value=42";
+	BOOST_CHECK_EQUAL(req[2].substr(0, perfData.length()), perfData);
+	PauseWriter();
+}
+
+BOOST_AUTO_TEST_CASE(pause_with_pending_work)
+{
+	ResumeWriter();
+
+	// Process check-results until the writer is stuck.
+	BOOST_REQUIRE_MESSAGE(GetWriterStuck(10s), "Failed to get Writer stuck.");
+
+	// Now try to pause.
+	PauseWriter();
+
+	REQUIRE_LOG_MESSAGE("Connection stopped\\.", 10s);
+	REQUIRE_LOG_MESSAGE("'Influxdb2Writer' paused\\.", 1s);
+}
+
+BOOST_AUTO_TEST_SUITE_END()

test/perfdata-opentsdbwriter.cpp

@@ -0,0 +1,53 @@
+// SPDX-FileCopyrightText: 2026 Icinga GmbH <https://icinga.com>
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include <BoostTestTargetConfig.h>
+#include "base/perfdatavalue.hpp"
+#include "perfdata/opentsdbwriter.hpp"
+#include "test/base-testloggerfixture.hpp"
+#include "test/perfdata-perfdatawriterfixture.hpp"
+#include "test/utils.hpp"
+
+using namespace icinga;
+
+BOOST_FIXTURE_TEST_SUITE(perfdata_opentsdbwriter, PerfdataWriterFixture<OpenTsdbWriter>,
+	*boost::unit_test::label("perfdata")
+	*boost::unit_test::label("network")
+)
+
+BOOST_AUTO_TEST_CASE(connect)
+{
+	ResumeWriter();
+
+	ReceiveCheckResults(1, ServiceState::ServiceCritical);
+
+	Accept();
+	auto msg = GetDataUntil('\n');
+	std::vector<std::string> splitMsg;
+	boost::split(splitMsg, msg, boost::is_any_of(" "));
+
+	// Just some basic sanity tests. It's not important to check if everything is entirely correct here.
+	BOOST_REQUIRE_EQUAL(splitMsg.size(), 5);
+	BOOST_REQUIRE_EQUAL(splitMsg[0], "put");
+	BOOST_REQUIRE_EQUAL(splitMsg[1], "icinga.host.state");
+	BOOST_REQUIRE_CLOSE(boost::lexical_cast<double>(splitMsg[2]), Utility::GetTime(), 1);
+	BOOST_REQUIRE_EQUAL(splitMsg[3], "1");
+	BOOST_REQUIRE_EQUAL(splitMsg[4], "host=h1");
+	PauseWriter();
+}
+
+BOOST_AUTO_TEST_CASE(pause_with_pending_work)
+{
+	ResumeWriter();
+
+	// Process check-results until the writer is stuck.
+	BOOST_REQUIRE_MESSAGE(GetWriterStuck(10s), "Failed to get Writer stuck.");
+
+	// Now stop reading and try to pause OpenTsdbWriter.
+	PauseWriter();
+
+	REQUIRE_LOG_MESSAGE("Connection stopped\\.", 10s);
+	REQUIRE_LOG_MESSAGE("'OpenTsdbWriter' paused\\.", 10s);
+}
+
+BOOST_AUTO_TEST_SUITE_END()

test/perfdata-perfdatatargetfixture.hpp

@@ -0,0 +1,198 @@
+// SPDX-FileCopyrightText: 2026 Icinga GmbH <https://icinga.com>
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#pragma once
+
+#include <BoostTestTargetConfig.h>
+#include "base/io-engine.hpp"
+#include "base/json.hpp"
+#include "base/tlsstream.hpp"
+#include <boost/algorithm/string/classification.hpp>
+#include <boost/algorithm/string/split.hpp>
+#include <boost/asio/read_until.hpp>
+#include <boost/asio/streambuf.hpp>
+#include <boost/asio/use_future.hpp>
+#include <boost/beast/http.hpp>
+#include <boost/beast/http/message.hpp>
+#include <boost/beast/http/parser.hpp>
+#include <boost/beast/http/string_body.hpp>
+
+namespace icinga {
+
+/**
+ * A fixture that provides methods to simulate a perfdata target
+ */
+class PerfdataWriterTargetFixture
+{
+public:
+	PerfdataWriterTargetFixture()
+		: icinga::PerfdataWriterTargetFixture(Shared<AsioTcpStream>::Make(IoEngine::Get().GetIoContext()))
+	{
+	}
+
+	explicit PerfdataWriterTargetFixture(const Shared<boost::asio::ssl::context>::Ptr& sslCtx)
+		: icinga::PerfdataWriterTargetFixture(Shared<AsioTlsStream>::Make(IoEngine::Get().GetIoContext(), *sslCtx))
+	{
+		m_SslContext = sslCtx;
+	}
+
+	explicit PerfdataWriterTargetFixture(AsioTlsOrTcpStream stream)
+		: m_Stream(std::move(stream)),
+		  m_Acceptor(
+			  IoEngine::Get().GetIoContext(),
+			  boost::asio::ip::tcp::endpoint{boost::asio::ip::address_v4::loopback(), 0}
+		  )
+	{
+	}
+
+	unsigned short GetPort() { return m_Acceptor.local_endpoint().port(); }
+
+	void Accept()
+	{
+		BOOST_REQUIRE_NO_THROW(
+			std::visit([&](auto& stream) { return m_Acceptor.accept(stream->lowest_layer()); }, m_Stream)
+		);
+	}
+
+	void Handshake()
+	{
+		BOOST_REQUIRE(std::holds_alternative<Shared<AsioTlsStream>::Ptr>(m_Stream));
+		using handshake_type = UnbufferedAsioTlsStream::handshake_type;
+		auto& stream = std::get<Shared<AsioTlsStream>::Ptr>(m_Stream);
+		BOOST_REQUIRE_NO_THROW(stream->next_layer().handshake(handshake_type::server));
+		BOOST_REQUIRE(stream->next_layer().IsVerifyOK());
+	}
+
+	void Shutdown()
+	{
+		BOOST_REQUIRE(std::holds_alternative<Shared<AsioTlsStream>::Ptr>(m_Stream));
+		auto& stream = std::get<Shared<AsioTlsStream>::Ptr>(m_Stream);
+		try {
+			stream->next_layer().shutdown();
+		} catch (const std::exception& ex) {
+			if (const auto* se = dynamic_cast<const boost::system::system_error*>(&ex);
+				!se || se->code() != boost::asio::error::eof) {
+				BOOST_FAIL("Exception in shutdown(): " << ex.what());
+			}
+		}
+
+		ResetStream();
+	}
+
+	void ResetStream()
+	{
+		if (std::holds_alternative<Shared<AsioTlsStream>::Ptr>(m_Stream)) {
+			m_Stream = Shared<AsioTlsStream>::Make(IoEngine::Get().GetIoContext(), *m_SslContext);
+		} else {
+			m_Stream = Shared<AsioTcpStream>::Make(IoEngine::Get().GetIoContext());
+		}
+	}
+
+	/**
+	 * Reads the HTTP request body from the stream, with an optional limit on the number of bytes to read.
+	 *
+	 * @param bytes The maximum number of bytes to read from the request body. If 0, there is no limit and the entire body will be read.
+	 *
+	 * @return The HTTP request read from the stream.
+	 */
+	boost::beast::http::request<boost::beast::http::string_body> GetRequest(std::size_t bytes = 0)
+	{
+		using namespace boost::beast;
+
+		boost::beast::flat_buffer buf;
+		if (bytes > 0) {
+			buf = boost::beast::flat_buffer{bytes};
+		}
+		boost::system::error_code ec;
+		http::request_parser<http::string_body> parser;
+		parser.body_limit(-1);
+		std::visit(
+			[&](auto& stream) {
+				http::read(*stream, buf, parser, ec);
+			},
+			m_Stream
+		);
+		if (bytes > 0) {
+			BOOST_REQUIRE_MESSAGE(
+				!ec || ec == http::error::buffer_overflow,
+				"Reading request body with a buffer limit of '" << bytes <<
+				"' should either succeed or fail with a buffer_overflow error, but got: " << ec.message()
+			);
+		} else {
+			BOOST_REQUIRE_MESSAGE(!ec, "Error while reading request body: " << ec.message());
+			BOOST_REQUIRE_MESSAGE(parser.is_done(), "Parser did not finish reading the request, but no error was set.");
+		}
+		return parser.release();
+	}
+
+	auto GetSplitRequestBody(char delim)
+	{
+		auto request = GetRequest();
+		std::vector<std::string> result{};
+		boost::split(result, request.body(), boost::is_any_of(std::string{delim}));
+		return result;
+	}
+
+	auto GetSplitDecodedRequestBody()
+	{
+		Array::Ptr result = new Array;
+		for (const auto& line : GetSplitRequestBody('\n')) {
+			if (!line.empty()) {
+				result->Add(JsonDecode(line));
+			}
+		}
+		return result;
+	}
+
+	template<typename T>
+	std::string GetDataUntil(T&& delim)
+	{
+		using namespace boost::asio::ip;
+
+		std::size_t delimLength{1};
+		if constexpr (!std::is_same_v<std::decay_t<T>, char>) {
+			delimLength = std::string_view{delim}.size();
+		}
+
+		boost::asio::streambuf buf;
+		boost::system::error_code ec;
+		auto bytesRead = std::visit(
+			[&](auto& stream) { return boost::asio::read_until(*stream, buf, std::forward<T>(delim), ec); }, m_Stream
+		);
+		BOOST_REQUIRE_MESSAGE(!ec, ec.message());
+
+		std::string ret{
+			boost::asio::buffers_begin(buf.data()), boost::asio::buffers_begin(buf.data()) + bytesRead - delimLength
+		};
+		buf.consume(bytesRead);
+
+		return ret;
+	}
+
+	void SendResponse(boost::beast::http::status status = boost::beast::http::status::ok)
+	{
+		using namespace boost::asio::ip;
+		using namespace boost::beast;
+
+		boost::system::error_code ec;
+		http::response<boost::beast::http::empty_body> response;
+		response.result(status);
+		response.prepare_payload();
+		std::visit(
+			[&](auto& stream) {
+				http::write(*stream, response, ec);
+				BOOST_REQUIRE_MESSAGE(!ec, ec.message());
+				stream->flush(ec);
+				BOOST_REQUIRE_MESSAGE(!ec, ec.message());
+			},
+			m_Stream
+		);
+	}
+
+private:
+	AsioTlsOrTcpStream m_Stream;
+	boost::asio::ip::tcp::acceptor m_Acceptor;
+	Shared<boost::asio::ssl::context>::Ptr m_SslContext;
+};
+
+} // namespace icinga

test/perfdata-perfdatawriterconnection.cpp

@@ -0,0 +1,335 @@
+// SPDX-FileCopyrightText: 2026 Icinga GmbH <https://icinga.com>
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "perfdata/perfdatawriterconnection.hpp"
+#include "test/perfdata-perfdatatargetfixture.hpp"
+#include "test/remote-certificate-fixture.hpp"
+#include "test/test-ctest.hpp"
+#include "test/test-thread.hpp"
+#include "test/utils.hpp"
+
+using namespace icinga;
+
+class TlsPerfdataWriterFixture : public CertificateFixture, public PerfdataWriterTargetFixture
+{
+public:
+	TlsPerfdataWriterFixture() : PerfdataWriterTargetFixture(MakeContext("server"))
+	{
+		m_PdwSslContext = MakeContext("client");
+
+		m_Conn = new PerfdataWriterConnection{"Test", "test", "127.0.0.1", std::to_string(GetPort()), m_PdwSslContext};
+	}
+
+	auto& GetConnection() { return *m_Conn; }
+
+private:
+	Shared<boost::asio::ssl::context>::Ptr MakeContext(const std::string& name)
+	{
+		auto testCert = EnsureCertFor(name);
+		return SetupSslContext(
+			testCert.crtFile,
+			testCert.keyFile,
+			m_CaCrtFile.string(),
+			"",
+			DEFAULT_TLS_CIPHERS,
+			DEFAULT_TLS_PROTOCOLMIN,
+			DebugInfo()
+		);
+	}
+
+	Shared<boost::asio::ssl::context>::Ptr m_PdwSslContext;
+	PerfdataWriterConnection::Ptr m_Conn;
+};
+
+BOOST_FIXTURE_TEST_SUITE(perfdata_connection, TlsPerfdataWriterFixture,
+	*CTestProperties("FIXTURES_REQUIRED ssl_certs")
+	*boost::unit_test::label("perfdata")
+	*boost::unit_test::label("network")
+)
+
+/* If there is no acceptor listening on the other side, connecting should fail.
+ */
+BOOST_AUTO_TEST_CASE(connection_refused)
+{
+	std::promise<void> p;
+	TestThread timeoutThread{[&]() {
+		auto f = p.get_future();
+		GetConnection().CancelAfterTimeout(f, 50ms);
+	}};
+
+	BOOST_REQUIRE_THROW(
+		GetConnection().Send(boost::asio::const_buffer{"foobar", 7}), PerfdataWriterConnection::Stopped
+	);
+
+	REQUIRE_JOINS_WITHIN(timeoutThread, 1s);
+}
+
+/* The PerfdataWriterConnection connects automatically when sending the first data.
+ * In case of http we also need to support disconnecting and reconnecting.
+ */
+BOOST_AUTO_TEST_CASE(ensure_connected)
+{
+	std::promise<void> disconnectedPromise;
+
+	TestThread mockTargetThread{[&]() {
+		Accept();
+		Handshake();
+		auto ret = GetDataUntil('\0');
+		Shutdown();
+		disconnectedPromise.get_future().get();
+		BOOST_REQUIRE_EQUAL(ret, "foobar");
+	}};
+
+	BOOST_REQUIRE_NO_THROW(GetConnection().Send(boost::asio::const_buffer{"foobar", 7}));
+	BOOST_REQUIRE_NO_THROW(GetConnection().Disconnect());
+	disconnectedPromise.set_value();
+
+	REQUIRE_JOINS_WITHIN(mockTargetThread, 1s);
+}
+
+/* Verify that data can still be sent while CancelAfterTimeout is waiting and the timeout
+ * can be aborted when all data has been sent successfully.
+ */
+BOOST_AUTO_TEST_CASE(finish_during_timeout)
+{
+	std::promise<void> p;
+
+	TestThread mockTargetThread{[&]() {
+		Accept();
+		Handshake();
+		auto ret = GetDataUntil('\0');
+		BOOST_REQUIRE_EQUAL(ret, "foobar");
+		ret = GetDataUntil('\0');
+		BOOST_REQUIRE_EQUAL(ret, "foobar");
+		// This is done here instead of the main thread after send, because we need to
+		// synchronize the asserts done in the timeoutThread after this point.
+		p.set_value();
+		Shutdown();
+	}};
+
+	GetConnection().Send(boost::asio::const_buffer{"foobar", 7});
+
+	TestThread timeoutThread{[&]() {
+		auto f = p.get_future();
+		GetConnection().CancelAfterTimeout(f, 50ms);
+		BOOST_REQUIRE(f.wait_for(0ms) == std::future_status::ready);
+		BOOST_REQUIRE(!GetConnection().IsConnected());
+	}};
+
+	GetConnection().Send(boost::asio::const_buffer{"foobar", 7});
+
+	REQUIRE_JOINS_WITHIN(timeoutThread, 1s);
+	REQUIRE_JOINS_WITHIN(mockTargetThread, 1s);
+}
+
+/* For the client, even a hanging server will accept the connection immediately, since it's done
+ * in the kernel. But in that case the TLS handshake will be stuck, so we need to verify that a
+ * handshake can be interrupted by CancelAfterTimeout().
+ */
+BOOST_AUTO_TEST_CASE(stuck_in_handshake)
+{
+	TestThread mockTargetThread{[&]() { Accept(); }};
+
+	std::promise<void> p;
+	TestThread timeoutThread{[&]() {
+		auto f = p.get_future();
+		GetConnection().CancelAfterTimeout(f, 50ms);
+		BOOST_REQUIRE(f.wait_for(0ms) == std::future_status::timeout);
+	}};
+
+	BOOST_REQUIRE_THROW(
+		GetConnection().Send(boost::asio::const_buffer{"foobar", 7}), PerfdataWriterConnection::Stopped
+	);
+
+	REQUIRE_JOINS_WITHIN(timeoutThread, 1s);
+	REQUIRE_JOINS_WITHIN(mockTargetThread, 1s);
+}
+
+/* When the disconnect timeout runs out while sending something to a slow or blocking server, we
+ * expect the send to be aborted after a timeout with an 'operation cancelled' exception, in
+ * order to not delay the shutdown of a perfdata writer indefinitely.
+ * No orderly TLS shutdown can be performed in this case, because the stream has been truncated.
+ * The server will need to handle this one on their own.
+ */
+BOOST_AUTO_TEST_CASE(stuck_sending)
+{
+	std::promise<void> shutdownPromise;
+	std::promise<void> dataReadPromise;
+
+	TestThread mockTargetThread{[&]() {
+		Accept();
+		Handshake();
+		auto ret = GetDataUntil("#");
+		BOOST_REQUIRE_EQUAL(ret, "foobar");
+		dataReadPromise.set_value();
+
+		// There's still a full buffer waiting to be read, but we're pretending to be dead and
+		// close the socket at this point.
+		shutdownPromise.get_future().get();
+		ResetStream();
+	}};
+
+	TestThread timeoutThread{[&]() {
+		// Synchronize with when mockTargetThread has read the initial data.
+		// This should especially help with timing on slow machines like the ARM GHA runners.
+		dataReadPromise.get_future().get();
+		BOOST_REQUIRE(GetConnection().IsConnected());
+		BOOST_REQUIRE_NO_THROW(GetConnection().Disconnect());
+		BOOST_REQUIRE(!GetConnection().IsConnected());
+	}};
+
+	// Allocate a large string that will fill the buffers on both sides of the connection, in
+	// order to make Send() block.
+	auto randomData = GetRandomString("foobar#", 4UL * 1024 * 1024);
+	auto buf = boost::asio::const_buffer{randomData.data(), randomData.size()};
+	BOOST_REQUIRE_THROW(GetConnection().Send(buf), PerfdataWriterConnection::Stopped);
+	shutdownPromise.set_value();
+
+	REQUIRE_JOINS_WITHIN(timeoutThread, 1s);
+	REQUIRE_JOINS_WITHIN(mockTargetThread, 1s);
+}
+
+/* This simulates a server that is stuck after receiving a HTTP request and before sending their
+ * response. Here, the simulated server is polite and still responds to a shutdown request, but
+ * in reality a server might not even do that. That case should be handled by our
+ * AsioTlsStream::GracefulDisconnect() function with an additional 10s timeout.
+ */
+BOOST_AUTO_TEST_CASE(stuck_reading_response)
+{
+	std::promise<void> shutdownPromise;
+	std::promise<void> requestReadPromise;
+
+	TestThread mockTargetThread{[&]() {
+		Accept();
+		Handshake();
+		auto ret = GetRequest();
+		BOOST_REQUIRE_EQUAL(ret.body(), "bar");
+		requestReadPromise.set_value();
+		// Do not send a response but react to the shutdown to be polite.
+		shutdownPromise.get_future().get();
+		Shutdown();
+	}};
+
+	TestThread timeoutThread{[&]() {
+		// Synchronize with after mockTargetThread has read the request
+		requestReadPromise.get_future().get();
+		BOOST_REQUIRE(GetConnection().IsConnected());
+		BOOST_REQUIRE_NO_THROW(GetConnection().Disconnect());
+		BOOST_REQUIRE(!GetConnection().IsConnected());
+	}};
+
+	boost::beast::http::request<boost::beast::http::string_body> request;
+	request.body() = "bar";
+	request.method(boost::beast::http::verb::get);
+	request.target("foo");
+	request.prepare_payload();
+	BOOST_REQUIRE_THROW(GetConnection().Send(request), PerfdataWriterConnection::Stopped);
+	shutdownPromise.set_value();
+
+	REQUIRE_JOINS_WITHIN(timeoutThread, 1s);
+	REQUIRE_JOINS_WITHIN(mockTargetThread, 1s);
+}
+
+/* This test simulates a server that closes the connection and reappears at a later time.
+ * PerfdataWriterConnection should detect the disconnect, catch the exception and attempt to
+ * reconnect without exiting Send().
+ */
+BOOST_AUTO_TEST_CASE(reconnect_failed)
+{
+	TestThread mockTargetThread{[&]() {
+		Accept();
+		Handshake();
+		auto ret = GetDataUntil("#");
+		BOOST_REQUIRE_EQUAL(ret, "foobar");
+
+		ResetStream();
+
+		Accept();
+		Handshake();
+
+		ret = GetDataUntil("#");
+		BOOST_REQUIRE_EQUAL(ret, "foobar");
+		ret = GetDataUntil("\n");
+
+		Shutdown();
+	}};
+
+	// Allocate a large string that will fill the buffers on both sides of the connection, in
+	// order to make Send() block.
+	auto randomData = GetRandomString("foobar#", 4UL * 1024 * 1024);
+	randomData.push_back('\n');
+	BOOST_REQUIRE_NO_THROW(GetConnection().Send(boost::asio::const_buffer{randomData.data(), randomData.size()}));
+	BOOST_REQUIRE_NO_THROW(GetConnection().Disconnect());
+
+	REQUIRE_JOINS_WITHIN(mockTargetThread, 1s);
+}
+
+/* This tests if retrying an http send will reproducibly lead to the exact same message being
+ * received. Normally this us guaranteed by the interface only accepting a const reference, but
+ * since on older boost versions the async_write() functions also accept non-const references, it
+ * doesn't hurt to ensure this with a test-case.
+ */
+BOOST_AUTO_TEST_CASE(http_send_retry)
+{
+	TestThread mockTargetThread{[&] {
+		Accept();
+		Handshake();
+
+		/* Read only the first 512 bytes of the request body, since we don't want to unblock the client yet.
+		 */
+		auto request = GetRequest(512);
+		BOOST_REQUIRE_MESSAGE(
+			request.method() == boost::beast::http::verb::post,
+			"Request method is not POST: " << request.method_string()
+		);
+		BOOST_REQUIRE_MESSAGE(request.target() == "foo", "Request target is not 'foo': " << request.target());
+		BOOST_REQUIRE_MESSAGE(
+			request.body().compare(0, 7, "foobar#") == 0,
+			"Request body does not start with 'foobar#': " << request.body().substr(0, 7)
+		);
+
+		ResetStream();
+		Accept();
+		Handshake();
+
+		/* Read the entire response now and verify that we still get the expected body,
+		 * even though the first read was only partial.
+		 */
+		request = GetRequest();
+		BOOST_REQUIRE_MESSAGE(
+			request.method() == boost::beast::http::verb::post,
+			"Request method is not POST: " << request.method_string()
+		);
+		BOOST_REQUIRE_MESSAGE(request.target() == "foo", "Request target is not 'foo': " << request.target());
+		BOOST_REQUIRE_MESSAGE(
+			request.body().compare(0, 7, "foobar#") == 0,
+			"Request body does not start with 'foobar#': " << request.body().substr(0, 7)
+		);
+
+		/* The body size is 4MB + 7 bytes (7 bytes for the "foobar#" prefix of the generated message)
+		 */
+		BOOST_REQUIRE_MESSAGE(
+			request.body().size() == (4UL * 1024 * 1024) + 7,
+			"Request body is not the expected size: " << request.body().size()
+		);
+
+		SendResponse();
+
+		Shutdown();
+	}};
+
+	boost::beast::http::request<boost::beast::http::string_body> request{boost::beast::http::verb::post, "foo", 10};
+	request.set(boost::beast::http::field::host, "localhost:" + std::to_string(GetPort()));
+
+	/* Allocate a large string that will fill the buffers on both sides of the connection, in
+	 * order to make Send() block.
+	 */
+	request.body() = GetRandomString("foobar#", 4UL * 1024 * 1024);
+	request.prepare_payload();
+	BOOST_REQUIRE_NO_THROW(GetConnection().Send(request));
+	BOOST_REQUIRE_NO_THROW(GetConnection().Disconnect());
+
+	REQUIRE_JOINS_WITHIN(mockTargetThread, 1s);
+}
+
+BOOST_AUTO_TEST_SUITE_END()

test/perfdata-perfdatawriterfixture.hpp

@@ -0,0 +1,139 @@
+// SPDX-FileCopyrightText: 2026 Icinga GmbH <https://icinga.com>
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#pragma once
+
+#include <BoostTestTargetConfig.h>
+#include "base/perfdatavalue.hpp"
+#include "config/configcompiler.hpp"
+#include "config/configitem.hpp"
+#include "icinga/host.hpp"
+#include "test/base-testloggerfixture.hpp"
+#include "test/perfdata-perfdatatargetfixture.hpp"
+#include "test/utils.hpp"
+#include <boost/hana.hpp>
+
+namespace icinga {
+
+template<typename Writer>
+class PerfdataWriterFixture : public PerfdataWriterTargetFixture, public TestLoggerFixture
+{
+public:
+	PerfdataWriterFixture() : m_Writer(new Writer)
+	{
+		auto createObjects = [&]() {
+			String config = R"CONFIG(
+object CheckCommand "dummy" {
+	command = "/bin/echo"
+}
+object Host "h1" {
+	address = "h1"
+	check_command = "dummy"
+	enable_notifications = true
+	enable_active_checks = false
+	enable_passive_checks = true
+}
+)CONFIG";
+
+			std::unique_ptr<Expression> expr = ConfigCompiler::CompileText("<test>", config);
+			expr->Evaluate(*ScriptFrame::GetCurrentFrame());
+		};
+
+		ConfigItem::RunWithActivationContext(new Function("CreateTestObjects", createObjects));
+
+		m_Host = Host::GetByName("h1");
+		BOOST_REQUIRE(m_Host);
+
+		m_Writer->SetPort(std::to_string(GetPort()));
+		m_Writer->SetName(m_Writer->GetReflectionType()->GetName());
+		m_Writer->SetDisconnectTimeout(0.05);
+		m_Writer->Register();
+
+		auto hasFlushInterval = boost::hana::is_valid([](auto&& obj) -> decltype(obj.SetFlushInterval(0.05)) {});
+		if constexpr (decltype(hasFlushInterval(std::declval<Writer>()))::value) {
+			m_Writer->SetFlushInterval(0.05);
+		}
+
+		auto hasFlushThreshold = boost::hana::is_valid([](auto&& obj) -> decltype(obj.SetFlushThreshold(1)) {});
+		if constexpr (decltype(hasFlushThreshold(std::declval<Writer>()))::value) {
+			m_Writer->SetFlushThreshold(1);
+		}
+	}
+
+	void ReceiveCheckResults(
+		std::size_t num,
+		ServiceState state,
+		const std::function<void(const CheckResult::Ptr&)>& fn = {}
+	)
+	{
+		::ReceiveCheckResults(m_Host, num, state, fn);
+	}
+
+	std::size_t GetWorkQueueLength()
+	{
+		Array::Ptr dummy = new Array;
+		Dictionary::Ptr status = new Dictionary;
+		m_Writer->StatsFunc(status, dummy);
+		ObjectLock lock{status};
+		// Unpack the single-key top-level dictionary
+		Dictionary::Ptr writer = status->Begin()->second;
+		BOOST_REQUIRE(writer);
+		Dictionary::Ptr values = writer->Get(m_Writer->GetName());
+		BOOST_REQUIRE(values);
+		BOOST_REQUIRE(values->Contains("work_queue_items"));
+		return values->Get("work_queue_items");
+	}
+
+	/**
+	 * Processes check results until the writer's work queue is no longer moving.
+	 *
+	 * @param timeout Time after which to give up trying to get the writer stuck
+	 * @return true if the writer is now stuck
+	 */
+	bool GetWriterStuck(std::chrono::milliseconds timeout)
+	{
+		auto start = std::chrono::steady_clock::now();
+		std::size_t unchangedCount = 0;
+		while(true){
+			ReceiveCheckResults(10, ServiceCritical, [&](const CheckResult::Ptr& cr) {
+				cr->GetPerformanceData()->Add(new PerfdataValue{GetRandomString("", 4096), 1});
+			});
+
+			if (std::chrono::steady_clock::now() - start >= timeout) {
+				return false;
+			}
+
+			auto numWq = GetWorkQueueLength();
+			if (numWq >= 10) {
+				std::this_thread::sleep_for(1ms);
+				if (numWq == GetWorkQueueLength()) {
+					if (unchangedCount < 5) {
+						++unchangedCount;
+						continue;
+					}
+					return true;
+				}
+
+				unchangedCount = 0;
+			}
+		}
+	}
+
+	void ResumeWriter()
+	{
+		static_cast<ConfigObject::Ptr>(m_Writer)->OnConfigLoaded();
+		m_Writer->SetActive(true);
+		m_Writer->Activate();
+		BOOST_REQUIRE(!m_Writer->IsPaused());
+	}
+
+	void PauseWriter() { static_cast<ConfigObject::Ptr>(m_Writer)->Pause(); }
+
+	auto GetWriter() { return m_Writer; }
+
+private:
+	Host::Ptr m_Host;
+	typename Writer::Ptr m_Writer;
+};
+
+} // namespace icinga

test/utils.cpp

@@ -5,6 +5,7 @@
 #include "base/perfdatavalue.hpp"
 #include <cstring>
 #include <iomanip>
+#include <random>
 #include <sstream>
 #include <boost/test/unit_test.hpp>
 
@@ -68,6 +69,19 @@ GlobalTimezoneFixture::~GlobalTimezoneFixture()
     tzset();
 }
 
+std::string GetRandomString(std::string prefix, std::size_t length)
+{
+	std::random_device rd;
+	std::mt19937 gen(rd());
+	std::uniform_int_distribution<int> distribution('!', '~');
+
+	for (auto i = 0U; i < length; i++) {
+		prefix += static_cast<char>(distribution(gen));
+	}
+
+	return prefix;
+}
+
 /**
  * Make our test host receive a number of check-results.
  *

test/utils.hpp

@@ -27,6 +27,8 @@ struct GlobalTimezoneFixture
     char *tz;
 };
 
+std::string GetRandomString(std::string prefix, std::size_t length);
+
 void ReceiveCheckResults(
 	const icinga::Checkable::Ptr& host,
 	std::size_t num,