Changes
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 8af2b29..4139c50 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -4,11 +4,21 @@ project(status_tracker LANGUAGES CXX)
include(FetchContent)
set(LIBMW_BUILD_SQLITE ON CACHE BOOL "Build libmw SQLite support")
+set(LIBMW_BUILD_URL ON)
FetchContent_Declare(libmw
GIT_REPOSITORY https://github.com/MetroWind/libmw.git
GIT_TAG HEAD
)
FetchContent_MakeAvailable(libmw)
+set(CARES_SHARED OFF CACHE BOOL "Build shared c-ares library")
+set(CARES_STATIC ON CACHE BOOL "Build static c-ares library")
+set(CARES_BUILD_TOOLS OFF CACHE BOOL "Build c-ares tools")
+FetchContent_Declare(cares
+ GIT_REPOSITORY https://github.com/c-ares/c-ares.git
+ GIT_TAG HEAD
+)
+FetchContent_MakeAvailable(cares)
+find_package(CURL REQUIRED)
find_package(SQLite3 REQUIRED)
find_package(Threads REQUIRED)
if(NOT TARGET SQLite3::SQLite3)
@@ -17,10 +27,17 @@ endif()
set(SOURCE_FILES
src/data_source_sqlite.cpp
+ src/probe.cpp
+ src/socket_probe.cpp
+ src/scheduler.cpp
+ src/service.cpp
src/thread_pool.cpp
)
set(LIBS
mw::sqlite
+ mw::url
+ CURL::libcurl
+ c-ares::cares
SQLite3::SQLite3
Threads::Threads
)
@@ -45,7 +62,12 @@ if(BUILD_TESTING)
include(GoogleTest)
set(TEST_FILES
+ src/fake_probe.cpp
tests/data_source_sqlite_test.cpp
+ tests/fake_probe_test.cpp
+ tests/probe_test.cpp
+ tests/scheduler_test.cpp
+ tests/service_test.cpp
tests/thread_pool_test.cpp
)
add_executable(status_tracker_test ${SOURCE_FILES} ${TEST_FILES})
diff --git a/src/configuration.h b/src/configuration.h
new file mode 100644
index 0000000..daa91b0
--- /dev/null
+++ b/src/configuration.h
@@ -0,0 +1,27 @@
+#pragma once
+
+#include <cstddef>
+#include <string>
+#include <vector>
+
+#include "service_config.h"
+
+/// A group of services displayed together in the web UI.
+struct ServiceGroupConfig
+{
+ /// Group name shown in the web UI.
+ std::string name;
+ /// Service definitions in configuration order.
+ std::vector<ServiceConfig> services;
+};
+
+/// Application settings represented independently of the configuration format.
+struct Configuration
+{
+ /// Positive number of probe worker threads.
+ std::size_t worker_count = 0;
+ /// Path to the SQLite database used for status history.
+ std::string database_path;
+ /// Service groups in configuration order.
+ std::vector<ServiceGroupConfig> groups;
+};
diff --git a/src/data_source_interface.h b/src/data_source_interface.h
index 1d06080..1635498 100644
--- a/src/data_source_interface.h
+++ b/src/data_source_interface.h
@@ -7,13 +7,7 @@
#include <mw/error.hpp>
-/// Persisted probe outcomes; missing or stale data is derived separately.
-enum class ProbeStatus : int
-{
- GOOD = 0,
- BAD = 1,
- OTHER = 2
-};
+#include "probe_status.h"
/// One recorded probe result, independent of endpoint configuration.
struct StatusRecord
diff --git a/src/fake_probe.cpp b/src/fake_probe.cpp
new file mode 100644
index 0000000..d10cf69
--- /dev/null
+++ b/src/fake_probe.cpp
@@ -0,0 +1,14 @@
+#include "fake_probe.h"
+
+#include <thread>
+
+FakeProbe::FakeProbe(ProbeResult result,
+ std::chrono::steady_clock::duration delay)
+ : result(result), delay(delay)
+{}
+
+mw::E<ProbeResult> FakeProbe::probe()
+{
+ std::this_thread::sleep_for(delay);
+ return result;
+}
diff --git a/src/fake_probe.h b/src/fake_probe.h
new file mode 100644
index 0000000..be1e8ec
--- /dev/null
+++ b/src/fake_probe.h
@@ -0,0 +1,18 @@
+#pragma once
+
+#include "probe.h"
+
+/// Test probe that returns a configured result after a delay.
+class FakeProbe final : public ProbeInterface
+{
+public:
+ /// Configure the result and the delay before returning it.
+ FakeProbe(ProbeResult result, std::chrono::steady_clock::duration delay);
+
+ /// Sleep for the configured delay and return the supplied result.
+ mw::E<ProbeResult> probe() override;
+
+private:
+ ProbeResult result;
+ std::chrono::steady_clock::duration delay;
+};
diff --git a/src/probe.cpp b/src/probe.cpp
new file mode 100644
index 0000000..33b99e0
--- /dev/null
+++ b/src/probe.cpp
@@ -0,0 +1,310 @@
+#include "probe.h"
+#include "socket_probe.h"
+
+#include <climits>
+#include <exception>
+#include <span>
+#include <utility>
+
+#include <ares.h>
+#include <curl/curl.h>
+#include <mw/http_client.hpp>
+#include <mw/url.hpp>
+
+namespace
+{
+
+using Clock = std::chrono::steady_clock;
+
+struct NetworkRuntime
+{
+ CURLcode curl_status = curl_global_init(CURL_GLOBAL_DEFAULT);
+ int dns_status = ares_library_init(ARES_LIB_INIT_ALL);
+
+ ~NetworkRuntime()
+ {
+ if(dns_status == ARES_SUCCESS)
+ {
+ ares_library_cleanup();
+ }
+ if(curl_status == CURLE_OK)
+ {
+ curl_global_cleanup();
+ }
+ }
+};
+
+bool validHost(const std::string& host)
+{
+ return !host.empty() && host.find('\0') == std::string::npos;
+}
+
+mw::E<void> validate(const HttpEndpoint& config)
+{
+ if(!validHost(config.url))
+ {
+ return std::unexpected(mw::runtimeError("Invalid HTTP URL"));
+ }
+ auto url = mw::URL::fromStr(config.url);
+ if(!url || (url->scheme() != "http" && url->scheme() != "https") ||
+ url->host().empty())
+ {
+ return std::unexpected(mw::runtimeError("Expected HTTP(S) URL"));
+ }
+ const auto features = curl_version_info(CURLVERSION_NOW)->features;
+ if(!(features & CURL_VERSION_ASYNCHDNS))
+ {
+ return std::unexpected(mw::runtimeError(
+ "HTTP probes require libcurl with asynchronous DNS"));
+ }
+ return {};
+}
+
+mw::E<void> validate(const TcpEndpoint& config)
+{
+ if(!validHost(config.host) || config.port == 0)
+ {
+ return std::unexpected(mw::runtimeError("Invalid TCP endpoint"));
+ }
+ return {};
+}
+
+mw::E<void> validate(const UdpEndpoint& config)
+{
+ if(!validHost(config.host) || config.port == 0 ||
+ config.payload.size() > 65507)
+ {
+ return std::unexpected(mw::runtimeError("Invalid UDP endpoint"));
+ }
+ return {};
+}
+
+mw::E<void> validate(const IcmpEndpoint& config)
+{
+ if(!validHost(config.host))
+ {
+ return std::unexpected(mw::runtimeError("Invalid ICMP endpoint"));
+ }
+ return {};
+}
+
+mw::E<void> validateTimeout(std::chrono::seconds timeout,
+ Clock::time_point start)
+{
+ const auto maximum = std::chrono::duration_cast<std::chrono::seconds>(
+ Clock::time_point::max() - start);
+ if(timeout.count() <= 0 || timeout > maximum || timeout.count() > LONG_MAX)
+ {
+ return std::unexpected(mw::runtimeError("Invalid probe timeout"));
+ }
+ return {};
+}
+
+mw::E<void> initializeNetworking()
+{
+ static const NetworkRuntime runtime;
+ if(runtime.curl_status != CURLE_OK || runtime.dns_status != ARES_SUCCESS)
+ {
+ return std::unexpected(mw::runtimeError(
+ "Failed to initialize networking libraries"));
+ }
+ return {};
+}
+
+template<typename Config>
+mw::E<void> prepare(const Config& config, std::chrono::seconds timeout)
+{
+ auto initialized = initializeNetworking();
+ if(!initialized)
+ {
+ return initialized;
+ }
+ auto valid_timeout = validateTimeout(timeout, Clock::now());
+ if(!valid_timeout)
+ {
+ return valid_timeout;
+ }
+ return validate(config);
+}
+
+bool discardBody([[maybe_unused]] std::span<const std::byte> chunk)
+{
+ return true;
+}
+
+mw::E<ProbeStatus> check(const HttpEndpoint& config,
+ std::chrono::seconds timeout,
+ [[maybe_unused]] Clock::time_point deadline)
+{
+ mw::HTTPSession session;
+ auto configured = session.transferTimeout(timeout);
+ if(!configured)
+ {
+ return std::unexpected(configured.error());
+ }
+ configured = session.connectionTimeout(timeout);
+ if(!configured)
+ {
+ return std::unexpected(configured.error());
+ }
+ configured = session.allowedProtocols("http,https");
+ if(!configured)
+ {
+ return std::unexpected(configured.error());
+ }
+ session.followRedirects(false);
+ auto response = session.getStream(mw::HTTPRequest(config.url), discardBody);
+ if(!response)
+ {
+ return ProbeStatus::BAD;
+ }
+ return response->status >= 200 && response->status < 300
+ ? ProbeStatus::GOOD : ProbeStatus::BAD;
+}
+
+template<typename Config>
+mw::E<ProbeStatus> check(const Config& config,
+ [[maybe_unused]] std::chrono::seconds timeout,
+ Clock::time_point deadline)
+{
+ return probe_internal::probeSocket(config, deadline);
+}
+
+template<typename Config>
+mw::E<ProbeResult> measure(const Config& config, std::chrono::seconds timeout)
+{
+ try
+ {
+ const auto start = Clock::now();
+ auto valid_timeout = validateTimeout(timeout, start);
+ if(!valid_timeout)
+ {
+ return std::unexpected(valid_timeout.error());
+ }
+ const auto timestamp = std::chrono::duration_cast<std::chrono::seconds>(
+ std::chrono::system_clock::now().time_since_epoch()).count();
+ auto status = check(config, timeout, start + timeout);
+ if(!status)
+ {
+ return std::unexpected(status.error());
+ }
+ const auto duration =
+ std::chrono::duration_cast<std::chrono::microseconds>(
+ Clock::now() - start).count();
+ return ProbeResult{*status, timestamp, duration};
+ }
+ catch(const std::exception& error)
+ {
+ return std::unexpected(mw::runtimeError(error.what()));
+ }
+}
+
+}
+
+HttpProbe::HttpProbe(HttpEndpoint config, std::chrono::seconds timeout)
+ : config(std::move(config)), timeout(timeout)
+{}
+
+mw::E<ProbeResult> HttpProbe::probe()
+{
+ return measure(config, timeout);
+}
+
+mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const HttpEndpoint& config, std::chrono::seconds timeout)
+{
+ try
+ {
+ auto valid = prepare(config, timeout);
+ if(!valid)
+ {
+ return std::unexpected(valid.error());
+ }
+ return std::unique_ptr<ProbeInterface>(new HttpProbe(config, timeout));
+ }
+ catch(const std::exception& error)
+ {
+ return std::unexpected(mw::runtimeError(error.what()));
+ }
+}
+
+TcpProbe::TcpProbe(TcpEndpoint config, std::chrono::seconds timeout)
+ : config(std::move(config)), timeout(timeout)
+{}
+
+mw::E<ProbeResult> TcpProbe::probe()
+{
+ return measure(config, timeout);
+}
+
+mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const TcpEndpoint& config, std::chrono::seconds timeout)
+{
+ try
+ {
+ auto valid = prepare(config, timeout);
+ if(!valid)
+ {
+ return std::unexpected(valid.error());
+ }
+ return std::unique_ptr<ProbeInterface>(new TcpProbe(config, timeout));
+ }
+ catch(const std::exception& error)
+ {
+ return std::unexpected(mw::runtimeError(error.what()));
+ }
+}
+
+UdpProbe::UdpProbe(UdpEndpoint config, std::chrono::seconds timeout)
+ : config(std::move(config)), timeout(timeout)
+{}
+
+mw::E<ProbeResult> UdpProbe::probe()
+{
+ return measure(config, timeout);
+}
+
+mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const UdpEndpoint& config, std::chrono::seconds timeout)
+{
+ try
+ {
+ auto valid = prepare(config, timeout);
+ if(!valid)
+ {
+ return std::unexpected(valid.error());
+ }
+ return std::unique_ptr<ProbeInterface>(new UdpProbe(config, timeout));
+ }
+ catch(const std::exception& error)
+ {
+ return std::unexpected(mw::runtimeError(error.what()));
+ }
+}
+
+IcmpProbe::IcmpProbe(IcmpEndpoint config, std::chrono::seconds timeout)
+ : config(std::move(config)), timeout(timeout)
+{}
+
+mw::E<ProbeResult> IcmpProbe::probe()
+{
+ return measure(config, timeout);
+}
+
+mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const IcmpEndpoint& config, std::chrono::seconds timeout)
+{
+ try
+ {
+ auto valid = prepare(config, timeout);
+ if(!valid)
+ {
+ return std::unexpected(valid.error());
+ }
+ return std::unique_ptr<ProbeInterface>(new IcmpProbe(config, timeout));
+ }
+ catch(const std::exception& error)
+ {
+ return std::unexpected(mw::runtimeError(error.what()));
+ }
+}
diff --git a/src/probe.h b/src/probe.h
new file mode 100644
index 0000000..4b7146b
--- /dev/null
+++ b/src/probe.h
@@ -0,0 +1,152 @@
+#pragma once
+
+#include <chrono>
+#include <cstdint>
+#include <string>
+#include <memory>
+
+#include <mw/error.hpp>
+
+#include "probe_status.h"
+
+/// HTTP configuration; only a direct 2xx response is healthy.
+struct HttpEndpoint
+{
+ /// HTTP or HTTPS URL, after applying any service URL fallback.
+ std::string url;
+};
+
+/// TCP connection configuration.
+struct TcpEndpoint
+{
+ /// Hostname or IPv4/IPv6 address.
+ std::string host;
+ /// Destination port, from 1 to 65535.
+ std::uint16_t port;
+};
+
+/// UDP configuration; silence is OTHER, not proof of an open port.
+struct UdpEndpoint
+{
+ /// Hostname or IPv4/IPv6 address.
+ std::string host;
+ /// Destination port, from 1 to 65535.
+ std::uint16_t port;
+ /// Datagram bytes; defaults to an empty datagram.
+ std::string payload;
+};
+
+/// ICMP echo configuration for Linux ping sockets.
+struct IcmpEndpoint
+{
+ /// Hostname or IPv4/IPv6 address.
+ std::string host;
+};
+
+/// Measured result ready to be associated with a service and persisted.
+struct ProbeResult
+{
+ /// Outcome of the network check.
+ ProbeStatus status;
+ /// Probe start time in UTC Unix seconds.
+ std::int64_t timestamp;
+ /// Elapsed monotonic time in microseconds, including DNS resolution.
+ std::int64_t duration_microsecond;
+};
+
+/// A task-owned probe for one endpoint check.
+class ProbeInterface
+{
+public:
+ /// Release the probe through its interface.
+ virtual ~ProbeInterface() = default;
+
+ /// Check the endpoint and measure its result.
+ /// Network failures return BAD; silent UDP endpoints return OTHER.
+ /// Local execution failures return errors.
+ virtual mw::E<ProbeResult> probe() = 0;
+};
+
+/// HTTP GET probe owning its endpoint configuration.
+class HttpProbe final : public ProbeInterface
+{
+public:
+ /// Perform the configured HTTP GET check.
+ mw::E<ProbeResult> probe() override;
+
+private:
+ HttpProbe(HttpEndpoint config, std::chrono::seconds timeout);
+ friend mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const HttpEndpoint& config, std::chrono::seconds timeout);
+
+ HttpEndpoint config;
+ std::chrono::seconds timeout;
+};
+
+/// TCP connection probe owning its endpoint configuration.
+class TcpProbe final : public ProbeInterface
+{
+public:
+ /// Perform the configured TCP connection check.
+ mw::E<ProbeResult> probe() override;
+
+private:
+ TcpProbe(TcpEndpoint config, std::chrono::seconds timeout);
+ friend mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const TcpEndpoint& config, std::chrono::seconds timeout);
+
+ TcpEndpoint config;
+ std::chrono::seconds timeout;
+};
+
+/// UDP request/reply probe owning its endpoint configuration.
+class UdpProbe final : public ProbeInterface
+{
+public:
+ /// Perform the configured UDP request/reply check.
+ mw::E<ProbeResult> probe() override;
+
+private:
+ UdpProbe(UdpEndpoint config, std::chrono::seconds timeout);
+ friend mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const UdpEndpoint& config, std::chrono::seconds timeout);
+
+ UdpEndpoint config;
+ std::chrono::seconds timeout;
+};
+
+/// ICMP echo probe owning its endpoint configuration.
+class IcmpProbe final : public ProbeInterface
+{
+public:
+ /// Perform the configured ICMP echo check.
+ mw::E<ProbeResult> probe() override;
+
+private:
+ IcmpProbe(IcmpEndpoint config, std::chrono::seconds timeout);
+ friend mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const IcmpEndpoint& config, std::chrono::seconds timeout);
+
+ IcmpEndpoint config;
+ std::chrono::seconds timeout;
+};
+
+/// Create a HTTP probe with a positive timeout for DNS and network I/O.
+mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const HttpEndpoint& config,
+ std::chrono::seconds timeout = std::chrono::seconds(5));
+
+/// Create a TCP probe with a positive timeout for DNS and network I/O.
+mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const TcpEndpoint& config,
+ std::chrono::seconds timeout = std::chrono::seconds(5));
+
+/// Create a UDP probe with a positive timeout for DNS and network I/O.
+mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const UdpEndpoint& config,
+ std::chrono::seconds timeout = std::chrono::seconds(5));
+
+/// Create a ICMP probe with a positive timeout for DNS and network I/O.
+mw::E<std::unique_ptr<ProbeInterface>> createProbe(
+ const IcmpEndpoint& config,
+ std::chrono::seconds timeout = std::chrono::seconds(5));
diff --git a/src/probe_status.h b/src/probe_status.h
new file mode 100644
index 0000000..ffcb59d
--- /dev/null
+++ b/src/probe_status.h
@@ -0,0 +1,9 @@
+#pragma once
+
+/// Persisted probe outcomes; missing or stale data is derived separately.
+enum class ProbeStatus : int
+{
+ GOOD = 0,
+ BAD = 1,
+ OTHER = 2
+};
diff --git a/src/scheduler.cpp b/src/scheduler.cpp
new file mode 100644
index 0000000..6608f20
--- /dev/null
+++ b/src/scheduler.cpp
@@ -0,0 +1,285 @@
+#include "scheduler.h"
+
+#include <exception>
+#include <utility>
+
+bool Scheduler::Later::operator()(const ScheduledTask& left,
+ const ScheduledTask& right) const
+{
+ if(left.due == right.due)
+ {
+ return left.service->config.id > right.service->config.id;
+ }
+ return left.due > right.due;
+}
+
+mw::E<Scheduler::Clock::time_point> Scheduler::nextDueTime(
+ Clock::time_point previous, Clock::duration interval, Clock::time_point now)
+{
+ if(interval <= Clock::duration::zero() ||
+ previous < Clock::time_point{} || now < previous)
+ {
+ return std::unexpected(mw::runtimeError("Invalid task schedule"));
+ }
+ const auto delay = interval - (now - previous) % interval;
+ if(delay > Clock::time_point::max() - now)
+ {
+ return std::unexpected(mw::runtimeError("Task due time overflow"));
+ }
+ return now + delay;
+}
+
+Scheduler::Scheduler(DataSourceInterface& data_source,
+ ProbeFactory probe_factory)
+ : data_source(data_source), probe_factory(std::move(probe_factory))
+{}
+
+mw::E<std::unique_ptr<Scheduler>> Scheduler::create(
+ const Configuration& configuration, DataSourceInterface& data_source,
+ ProbeFactory probe_factory)
+{
+ if(configuration.worker_count == 0 || !probe_factory)
+ {
+ return std::unexpected(mw::runtimeError(
+ "Scheduler requires workers and a probe factory"));
+ }
+ try
+ {
+ auto scheduler = std::unique_ptr<Scheduler>(
+ new Scheduler(data_source, std::move(probe_factory)));
+ for(const auto& group : configuration.groups)
+ {
+ for(const auto& config : group.services)
+ {
+ if(config.id.empty() ||
+ config.id.find('\0') != std::string::npos)
+ {
+ return std::unexpected(mw::runtimeError(
+ "Invalid service ID"));
+ }
+ if(scheduler->services.contains(config.id))
+ {
+ return std::unexpected(mw::runtimeError(
+ "Duplicate service ID: " + config.id));
+ }
+ const auto now = Clock::now();
+ auto next = nextDueTime(now, config.interval, now);
+ if(!next)
+ {
+ return std::unexpected(mw::runtimeError(
+ config.id + ": " + next.error().msg()));
+ }
+ auto service = std::make_unique<Service>();
+ service->config = config;
+ // Validate without running the probe or the injected factory.
+ auto probe = service->createProbe();
+ if(!probe)
+ {
+ return std::unexpected(mw::runtimeError(
+ config.id + ": " + probe.error().msg()));
+ }
+ scheduler->services.emplace(config.id, std::move(service));
+ }
+ }
+ auto pool = ThreadPool::create(configuration.worker_count);
+ if(!pool)
+ {
+ return std::unexpected(pool.error());
+ }
+ scheduler->pool = std::move(*pool);
+ return scheduler;
+ }
+ catch(const std::exception& error)
+ {
+ return std::unexpected(mw::runtimeError(error.what()));
+ }
+}
+
+const Service* Scheduler::findService(const std::string& service_id) const
+{
+ const auto found = services.find(service_id);
+ return found == services.end() ? nullptr : found->second.get();
+}
+
+mw::E<void> Scheduler::run(std::stop_token stop_token)
+{
+ {
+ std::lock_guard lock(mutex);
+ if(started)
+ {
+ return std::unexpected(mw::runtimeError(
+ "Scheduler has already started"));
+ }
+ started = true;
+ }
+ mw::E<void> result;
+ try
+ {
+ result = runLoop(stop_token);
+ }
+ catch(const std::exception& exception)
+ {
+ result = std::unexpected(mw::runtimeError(exception.what()));
+ }
+ pool->waitIdle();
+ std::lock_guard lock(mutex);
+ if(result && error)
+ {
+ return std::unexpected(*error);
+ }
+ return result;
+}
+
+mw::E<void> Scheduler::runLoop(std::stop_token stop_token)
+{
+ const auto initial_due = Clock::now();
+ for(const auto& entry : services)
+ {
+ queue.push({entry.second.get(), initial_due});
+ }
+ while(!stop_token.stop_requested())
+ {
+ std::unique_lock lock(mutex);
+ if(error)
+ {
+ return std::unexpected(*error);
+ }
+ const auto now = Clock::now();
+ if(!queue.empty() && queue.top().due <= now)
+ {
+ auto scheduled = queue.top();
+ auto& service = *scheduled.service;
+ auto next = nextDueTime(
+ scheduled.due, service.config.interval, now);
+ if(!next)
+ {
+ return std::unexpected(next.error());
+ }
+ queue.pop();
+ scheduled.due = *next;
+ queue.push(scheduled);
+ lock.unlock();
+ auto dispatched = dispatch(service);
+ if(!dispatched)
+ {
+ return std::unexpected(dispatched.error());
+ }
+ continue;
+ }
+ if(!queue.empty())
+ {
+ const auto next = queue.top().due;
+ wake.wait_until(lock, stop_token, next,
+ std::bind_front(&Scheduler::hasError, this));
+ }
+ else
+ {
+ wake.wait(lock, stop_token,
+ std::bind_front(&Scheduler::hasError, this));
+ }
+ }
+ return {};
+}
+
+namespace
+{
+
+struct FlightGuard
+{
+ std::atomic<bool>& in_flight;
+ bool active = true;
+
+ ~FlightGuard()
+ {
+ if(active)
+ {
+ in_flight = false;
+ }
+ }
+};
+
+}
+
+mw::E<void> Scheduler::dispatch(Service& service)
+{
+ if(service.in_flight.exchange(true))
+ {
+ return {};
+ }
+ FlightGuard guard{service.in_flight};
+ auto submitted = pool->trySubmit(Task{*this, service, nullptr});
+ if(!submitted)
+ {
+ return std::unexpected(submitted.error());
+ }
+ if(*submitted)
+ {
+ // The accepted task now clears the flag after probing and persistence.
+ guard.active = false;
+ }
+ return {};
+}
+
+void Scheduler::Task::operator()()
+{
+ scheduler.execute(*this);
+}
+
+void Scheduler::execute(Task& task)
+{
+ FlightGuard guard{task.service.in_flight};
+ const auto& service_id = task.service.config.id;
+ try
+ {
+ auto probe = probe_factory(task.service);
+ if(!probe)
+ {
+ reportError(service_id, std::move(probe.error()));
+ return;
+ }
+ if(!*probe)
+ {
+ reportError(service_id, mw::runtimeError(
+ "Probe factory returned a null probe"));
+ return;
+ }
+ task.probe = std::move(*probe);
+ auto result = task.probe->probe();
+ if(!result)
+ {
+ reportError(service_id, std::move(result.error()));
+ return;
+ }
+ auto saved = data_source.save({service_id, result->timestamp,
+ result->duration_microsecond, result->status});
+ if(!saved)
+ {
+ reportError(service_id, std::move(saved.error()));
+ }
+ }
+ catch(const std::exception& exception)
+ {
+ reportError(service_id, mw::runtimeError(exception.what()));
+ }
+ catch(...)
+ {
+ reportError(service_id, mw::runtimeError("Unknown task exception"));
+ }
+}
+
+void Scheduler::reportError(const std::string& service_id, mw::Error failure)
+{
+ {
+ std::lock_guard lock(mutex);
+ if(!error)
+ {
+ error = mw::runtimeError(service_id + ": " + failure.msg());
+ }
+ }
+ wake.notify_one();
+}
+
+bool Scheduler::hasError() const
+{
+ return error.has_value();
+}
diff --git a/src/scheduler.h b/src/scheduler.h
new file mode 100644
index 0000000..3f38c1f
--- /dev/null
+++ b/src/scheduler.h
@@ -0,0 +1,95 @@
+#pragma once
+
+#include <chrono>
+#include <condition_variable>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <queue>
+#include <stop_token>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "configuration.h"
+#include "data_source_interface.h"
+#include "service.h"
+#include "thread_pool.h"
+
+/// Dispatch recurring endpoint tasks to a fixed-capacity worker pool.
+class Scheduler
+{
+public:
+ /// Monotonic clock used for probe scheduling.
+ using Clock = std::chrono::steady_clock;
+
+ /// Factory called by worker tasks to create their probes. Custom factories
+ /// must support concurrent calls for different services.
+ using ProbeFactory = std::function<
+ mw::E<std::unique_ptr<ProbeInterface>>(const Service&)>;
+
+ /// Copy configured services and create workers. The data source must
+ /// outlive the scheduler and support concurrent saves. Duplicate IDs and
+ /// invalid settings return errors.
+ static mw::E<std::unique_ptr<Scheduler>> create(
+ const Configuration& configuration, DataSourceInterface& data_source,
+ ProbeFactory probe_factory = &Service::createProbe);
+
+ /// Find a service by ID, or return nullptr. The pointer stays valid until
+ /// scheduler destruction; its configuration is immutable after creation.
+ const Service* findService(const std::string& service_id) const;
+
+ /// Release workers. run() must have returned before destruction.
+ ~Scheduler() = default;
+
+ /// Run once on the calling thread, initially making every task due.
+ /// Skip occurrences while busy and preserve each task's interval cadence.
+ /// Stop on cancellation or execution error, then finish accepted work.
+ mw::E<void> run(std::stop_token stop_token);
+
+ /// Find the next interval boundary strictly after now, without catching up.
+ static mw::E<Clock::time_point> nextDueTime(
+ Clock::time_point previous, Clock::duration interval,
+ Clock::time_point now);
+
+private:
+ struct Task
+ {
+ Scheduler& scheduler;
+ Service& service;
+ std::unique_ptr<ProbeInterface> probe;
+
+ void operator()();
+ };
+
+ struct ScheduledTask
+ {
+ Service* service;
+ Clock::time_point due;
+ };
+
+ struct Later
+ {
+ bool operator()(const ScheduledTask& left,
+ const ScheduledTask& right) const;
+ };
+
+ Scheduler(DataSourceInterface& data_source, ProbeFactory probe_factory);
+ mw::E<void> runLoop(std::stop_token stop_token);
+ mw::E<void> dispatch(Service& service);
+ void execute(Task& task);
+ void reportError(const std::string& service_id, mw::Error error);
+ bool hasError() const;
+
+ std::mutex mutex;
+ std::condition_variable_any wake;
+ bool started = false;
+ std::optional<mw::Error> error;
+ std::priority_queue<ScheduledTask, std::vector<ScheduledTask>, Later> queue;
+ std::unordered_map<std::string, std::unique_ptr<Service>> services;
+ DataSourceInterface& data_source;
+ ProbeFactory probe_factory;
+ // Destroy workers before the task state they reference.
+ std::unique_ptr<ThreadPool> pool;
+};
diff --git a/src/service.cpp b/src/service.cpp
new file mode 100644
index 0000000..bb91825
--- /dev/null
+++ b/src/service.cpp
@@ -0,0 +1,44 @@
+#include "service.h"
+
+#include <exception>
+
+namespace
+{
+
+struct ProbeFactory
+{
+ std::chrono::seconds timeout;
+ const std::optional<std::string>& url;
+
+ mw::E<std::unique_ptr<ProbeInterface>> operator()(
+ const HttpEndpoint& endpoint) const
+ {
+ if(endpoint.url.empty() && url)
+ {
+ return ::createProbe(HttpEndpoint{*url}, timeout);
+ }
+ return ::createProbe(endpoint, timeout);
+ }
+
+ template<typename Endpoint>
+ mw::E<std::unique_ptr<ProbeInterface>> operator()(
+ const Endpoint& endpoint) const
+ {
+ return ::createProbe(endpoint, timeout);
+ }
+};
+
+}
+
+mw::E<std::unique_ptr<ProbeInterface>> Service::createProbe() const
+{
+ try
+ {
+ return std::visit(ProbeFactory{config.timeout, config.url},
+ config.endpoint);
+ }
+ catch(const std::exception& error)
+ {
+ return std::unexpected(mw::runtimeError(error.what()));
+ }
+}
diff --git a/src/service.h b/src/service.h
new file mode 100644
index 0000000..bbed8d5
--- /dev/null
+++ b/src/service.h
@@ -0,0 +1,18 @@
+#pragma once
+
+#include <atomic>
+
+#include "service_config.h"
+
+/// Long-lived monitored service. Configuration must remain unchanged while
+/// tasks use it, and the service must outlive those tasks.
+struct Service
+{
+ /// Settings owned by this runtime service.
+ ServiceConfig config;
+ /// Set when a task is dispatched and cleared when that task finishes.
+ std::atomic<bool> in_flight{false};
+
+ /// Create a fresh task-owned probe without changing in_flight.
+ mw::E<std::unique_ptr<ProbeInterface>> createProbe() const;
+};
diff --git a/src/service_config.h b/src/service_config.h
new file mode 100644
index 0000000..c0c4870
--- /dev/null
+++ b/src/service_config.h
@@ -0,0 +1,31 @@
+#pragma once
+
+#include <chrono>
+#include <optional>
+#include <string>
+#include <variant>
+
+#include "probe.h"
+
+/// Protocol-specific configuration for a monitored service.
+using EndpointConfig = std::variant<HttpEndpoint, TcpEndpoint, UdpEndpoint,
+ IcmpEndpoint>;
+
+/// Configured settings for one monitored service, without runtime state.
+struct ServiceConfig
+{
+ /// Unique service ID across all groups.
+ std::string id;
+ /// Display name shown in the web UI.
+ std::string name;
+ /// Description shown in the web UI.
+ std::string description;
+ /// Protocol and destination used to check this service.
+ EndpointConfig endpoint;
+ /// Positive timeout covering DNS resolution and network I/O.
+ std::chrono::seconds timeout;
+ /// Positive interval between scheduled probes.
+ std::chrono::steady_clock::duration interval;
+ /// Optional user-facing URL, also used as the HTTP endpoint fallback.
+ std::optional<std::string> url;
+};
diff --git a/src/socket_probe.cpp b/src/socket_probe.cpp
new file mode 100644
index 0000000..1c472ca
--- /dev/null
+++ b/src/socket_probe.cpp
@@ -0,0 +1,411 @@
+#include "socket_probe.h"
+
+#include <algorithm>
+#include <array>
+#include <atomic>
+#include <cerrno>
+#include <climits>
+#include <cstring>
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include <ares.h>
+#include <arpa/inet.h>
+#include <netinet/icmp6.h>
+#include <netinet/ip_icmp.h>
+#include <poll.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+namespace
+{
+
+using Clock = std::chrono::steady_clock;
+using Addresses = std::unique_ptr<ares_addrinfo, decltype(&ares_freeaddrinfo)>;
+
+enum class Protocol { TCP, UDP, ICMP };
+enum class Phase { CONNECTING, WRITING, READING, FAILED };
+
+struct Socket
+{
+ int fd;
+
+ explicit Socket(int fd) : fd(fd) {}
+ Socket(const Socket&) = delete;
+ Socket& operator=(const Socket&) = delete;
+ Socket(Socket&& other) noexcept : fd(std::exchange(other.fd, -1)) {}
+
+ ~Socket()
+ {
+ if(fd >= 0)
+ {
+ close(fd);
+ }
+ }
+};
+
+struct Resolution
+{
+ int status = ARES_ETIMEOUT;
+ Addresses addresses{nullptr, ares_freeaddrinfo};
+};
+
+void resolved(void* context, int status, [[maybe_unused]] int timeouts,
+ ares_addrinfo* addresses)
+{
+ auto& result = *static_cast<Resolution*>(context);
+ result.status = status;
+ result.addresses.reset(addresses);
+}
+
+int remainingMilliseconds(Clock::time_point deadline)
+{
+ const auto remaining = std::chrono::ceil<std::chrono::milliseconds>(
+ deadline - Clock::now()).count();
+ return static_cast<int>(std::clamp<std::int64_t>(remaining, 0, INT_MAX));
+}
+
+mw::E<Addresses> resolve(const std::string& host, std::uint16_t port,
+ Clock::time_point deadline)
+{
+ Resolution result;
+ ares_channel_t* raw = nullptr;
+ ares_options options{};
+ options.evsys = ARES_EVSYS_DEFAULT;
+ const int initialized = ares_init_options(&raw, &options,
+ ARES_OPT_EVENT_THREAD);
+ if(initialized != ARES_SUCCESS)
+ {
+ return std::unexpected(mw::runtimeError(ares_strerror(initialized)));
+ }
+ std::unique_ptr<ares_channel_t, decltype(&ares_destroy)> channel(
+ raw, ares_destroy);
+ ares_addrinfo_hints hints{};
+ hints.ai_family = AF_UNSPEC;
+ hints.ai_socktype = SOCK_DGRAM;
+ hints.ai_flags = ARES_AI_NUMERICSERV | ARES_AI_NOSORT;
+ const auto service = std::to_string(port);
+ ares_getaddrinfo(channel.get(), host.c_str(), service.c_str(), &hints,
+ resolved, &result);
+ const auto waited = ares_queue_wait_empty(channel.get(),
+ remainingMilliseconds(deadline));
+ // Destruction joins the resolver thread before we inspect callback state.
+ channel.reset();
+ if(waited != ARES_SUCCESS || result.status != ARES_SUCCESS)
+ {
+ if(result.status == ARES_ENOMEM)
+ {
+ return std::unexpected(mw::runtimeError("DNS allocation failed"));
+ }
+ return Addresses(nullptr, ares_freeaddrinfo);
+ }
+ return std::move(result.addresses);
+}
+
+mw::Error socketError(const char* operation)
+{
+ return mw::runtimeError(std::string(operation) + ": " +
+ std::strerror(errno));
+}
+
+bool retryable(int code)
+{
+ return code == EAGAIN || code == EWOULDBLOCK || code == EINTR;
+}
+
+bool networkFailure(int code)
+{
+ return code == ECONNREFUSED || code == ECONNRESET || code == ETIMEDOUT ||
+ code == EHOSTUNREACH || code == ENETUNREACH || code == ENETDOWN ||
+ code == EHOSTDOWN || code == EPIPE;
+}
+
+struct Connection
+{
+ Socket socket;
+ const ares_addrinfo_node* address;
+ Phase phase;
+ std::array<unsigned char, 16> echo{};
+};
+
+std::array<unsigned char, 16> echoRequest(int family)
+{
+ static std::atomic<std::uint64_t> next_nonce{1};
+ const auto nonce = next_nonce.fetch_add(1);
+ std::array<unsigned char, 16> packet{};
+ packet[0] = family == AF_INET ? ICMP_ECHO : ICMP6_ECHO_REQUEST;
+ packet[7] = 1;
+ for(std::size_t i = 0; i < sizeof(nonce); ++i)
+ {
+ packet[8 + i] = static_cast<unsigned char>(nonce >> (i * 8));
+ }
+ if(family == AF_INET)
+ {
+ unsigned int sum = 0;
+ for(std::size_t i = 0; i < packet.size(); i += 2)
+ {
+ sum += (packet[i] << 8) | packet[i + 1];
+ }
+ while(sum >> 16)
+ {
+ sum = (sum & 0xffff) + (sum >> 16);
+ }
+ const auto checksum = static_cast<std::uint16_t>(~sum);
+ packet[2] = checksum >> 8;
+ packet[3] = checksum & 0xff;
+ }
+ return packet;
+}
+
+bool sameHost(const sockaddr_storage& source, const sockaddr* expected)
+{
+ if(source.ss_family != expected->sa_family)
+ {
+ return false;
+ }
+ if(source.ss_family == AF_INET)
+ {
+ const auto& actual = reinterpret_cast<const sockaddr_in&>(source);
+ const auto* wanted = reinterpret_cast<const sockaddr_in*>(expected);
+ return actual.sin_addr.s_addr == wanted->sin_addr.s_addr;
+ }
+ const auto& actual = reinterpret_cast<const sockaddr_in6&>(source);
+ const auto* wanted = reinterpret_cast<const sockaddr_in6*>(expected);
+ return std::memcmp(&actual.sin6_addr, &wanted->sin6_addr,
+ sizeof(in6_addr)) == 0 &&
+ (wanted->sin6_scope_id == 0 ||
+ actual.sin6_scope_id == wanted->sin6_scope_id);
+}
+
+bool echoReply(const Connection& connection, const unsigned char* data,
+ std::size_t size, const sockaddr_storage& source)
+{
+ const int type = connection.address->ai_family == AF_INET
+ ? ICMP_ECHOREPLY : ICMP6_ECHO_REPLY;
+ return size == connection.echo.size() && data[0] == type && data[1] == 0 &&
+ data[6] == 0 && data[7] == 1 &&
+ std::memcmp(data + 8, connection.echo.data() + 8, 8) == 0 &&
+ sameHost(source, connection.address->ai_addr);
+}
+
+mw::E<ProbeStatus> check(const std::string& host, std::uint16_t port,
+ Protocol protocol, const std::string& payload,
+ Clock::time_point deadline)
+{
+ auto resolved = resolve(host, port, deadline);
+ if(!resolved)
+ {
+ return std::unexpected(resolved.error());
+ }
+ if(!*resolved || Clock::now() >= deadline)
+ {
+ return ProbeStatus::BAD;
+ }
+ auto addresses = std::move(*resolved);
+ std::vector<Connection> connections;
+ std::optional<mw::Error> setup_error;
+ bool attempted = false;
+ for(auto* address = addresses->nodes; address; address = address->ai_next)
+ {
+ if(address->ai_family != AF_INET && address->ai_family != AF_INET6)
+ {
+ continue;
+ }
+ const int type = protocol == Protocol::TCP ? SOCK_STREAM : SOCK_DGRAM;
+ const int ip_protocol = protocol == Protocol::ICMP
+ ? (address->ai_family == AF_INET ? static_cast<int>(IPPROTO_ICMP)
+ : static_cast<int>(IPPROTO_ICMPV6))
+ : 0;
+ Socket socket(::socket(address->ai_family,
+ type | SOCK_NONBLOCK | SOCK_CLOEXEC,
+ ip_protocol));
+ if(socket.fd < 0)
+ {
+ setup_error = socketError("Create probe socket");
+ continue;
+ }
+ attempted = true;
+ Phase phase = Phase::WRITING;
+ if(connect(socket.fd, address->ai_addr, address->ai_addrlen) < 0)
+ {
+ if(errno == EINPROGRESS || errno == EINTR)
+ {
+ phase = Phase::CONNECTING;
+ }
+ else if(networkFailure(errno))
+ {
+ continue;
+ }
+ else
+ {
+ return std::unexpected(socketError("Connect probe socket"));
+ }
+ }
+ else if(protocol == Protocol::TCP)
+ {
+ return ProbeStatus::GOOD;
+ }
+ connections.push_back({std::move(socket), address, phase,
+ echoRequest(address->ai_family)});
+ }
+ if(!attempted && setup_error)
+ {
+ return std::unexpected(*setup_error);
+ }
+ std::vector<pollfd> descriptors(connections.size());
+ while(Clock::now() < deadline)
+ {
+ bool active = false;
+ for(std::size_t i = 0; i < connections.size(); ++i)
+ {
+ const auto& connection = connections[i];
+ const bool failed = connection.phase == Phase::FAILED;
+ descriptors[i] = {failed ? -1 : connection.socket.fd,
+ static_cast<short>(connection.phase == Phase::READING
+ ? POLLIN : POLLOUT), 0};
+ active |= !failed;
+ }
+ if(!active)
+ {
+ return ProbeStatus::BAD;
+ }
+ const int ready = poll(descriptors.data(), descriptors.size(),
+ remainingMilliseconds(deadline));
+ if(ready < 0)
+ {
+ if(errno == EINTR)
+ {
+ continue;
+ }
+ return std::unexpected(socketError("Wait for probe socket"));
+ }
+ for(std::size_t i = 0; i < connections.size(); ++i)
+ {
+ if(!descriptors[i].revents)
+ {
+ continue;
+ }
+ auto& connection = connections[i];
+ if(descriptors[i].revents & POLLNVAL)
+ {
+ return std::unexpected(mw::runtimeError(
+ "Invalid probe socket"));
+ }
+ if(connection.phase == Phase::CONNECTING)
+ {
+ int code = 0;
+ socklen_t size = sizeof(code);
+ if(getsockopt(connection.socket.fd, SOL_SOCKET, SO_ERROR,
+ &code, &size) < 0)
+ {
+ return std::unexpected(socketError(
+ "Read connection error"));
+ }
+ if(code)
+ {
+ if(!networkFailure(code))
+ {
+ return std::unexpected(mw::runtimeError(
+ std::string("Connect: ") + std::strerror(code)));
+ }
+ connection.phase = Phase::FAILED;
+ continue;
+ }
+ if(protocol == Protocol::TCP)
+ {
+ return ProbeStatus::GOOD;
+ }
+ connection.phase = Phase::WRITING;
+ }
+ if(connection.phase == Phase::WRITING)
+ {
+ const void* data = protocol == Protocol::ICMP
+ ? static_cast<const void*>(connection.echo.data())
+ : static_cast<const void*>(payload.data());
+ const auto size = protocol == Protocol::ICMP
+ ? connection.echo.size() : payload.size();
+ const auto sent = send(connection.socket.fd, data, size,
+ MSG_NOSIGNAL);
+ if(sent < 0)
+ {
+ if(retryable(errno))
+ {
+ continue;
+ }
+ if(!networkFailure(errno))
+ {
+ return std::unexpected(socketError("Send probe"));
+ }
+ connection.phase = Phase::FAILED;
+ continue;
+ }
+ connection.phase = Phase::READING;
+ }
+ if(connection.phase == Phase::READING)
+ {
+ std::array<unsigned char, 65536> buffer;
+ sockaddr_storage source{};
+ socklen_t source_size = sizeof(source);
+ const auto received = recvfrom(connection.socket.fd,
+ buffer.data(), buffer.size(), 0,
+ reinterpret_cast<sockaddr*>(&source), &source_size);
+ if(received < 0)
+ {
+ if(retryable(errno))
+ {
+ continue;
+ }
+ if(!networkFailure(errno))
+ {
+ return std::unexpected(socketError("Receive probe"));
+ }
+ connection.phase = Phase::FAILED;
+ continue;
+ }
+ if(protocol == Protocol::UDP || echoReply(connection,
+ buffer.data(), static_cast<std::size_t>(received), source))
+ {
+ return ProbeStatus::GOOD;
+ }
+ }
+ }
+ }
+ if(protocol == Protocol::UDP)
+ {
+ for(const auto& connection : connections)
+ {
+ if(connection.phase == Phase::READING)
+ {
+ return ProbeStatus::OTHER;
+ }
+ }
+ }
+ return ProbeStatus::BAD;
+}
+
+}
+
+namespace probe_internal
+{
+
+mw::E<ProbeStatus> probeSocket(const TcpEndpoint& endpoint,
+ Clock::time_point deadline)
+{
+ return check(endpoint.host, endpoint.port, Protocol::TCP, {}, deadline);
+}
+
+mw::E<ProbeStatus> probeSocket(const UdpEndpoint& endpoint,
+ Clock::time_point deadline)
+{
+ return check(endpoint.host, endpoint.port, Protocol::UDP, endpoint.payload,
+ deadline);
+}
+
+mw::E<ProbeStatus> probeSocket(const IcmpEndpoint& endpoint,
+ Clock::time_point deadline)
+{
+ return check(endpoint.host, 0, Protocol::ICMP, {}, deadline);
+}
+
+}
diff --git a/src/socket_probe.h b/src/socket_probe.h
new file mode 100644
index 0000000..1cf13c3
--- /dev/null
+++ b/src/socket_probe.h
@@ -0,0 +1,18 @@
+#pragma once
+
+#include "probe.h"
+
+namespace probe_internal
+{
+
+/// Check a TCP endpoint before the shared monotonic deadline.
+mw::E<ProbeStatus> probeSocket(const TcpEndpoint& endpoint,
+ std::chrono::steady_clock::time_point deadline);
+/// Check a UDP endpoint before the shared monotonic deadline.
+mw::E<ProbeStatus> probeSocket(const UdpEndpoint& endpoint,
+ std::chrono::steady_clock::time_point deadline);
+/// Check an ICMP endpoint before the shared monotonic deadline.
+mw::E<ProbeStatus> probeSocket(const IcmpEndpoint& endpoint,
+ std::chrono::steady_clock::time_point deadline);
+
+}
diff --git a/tests/fake_probe_test.cpp b/tests/fake_probe_test.cpp
new file mode 100644
index 0000000..8c18cce
--- /dev/null
+++ b/tests/fake_probe_test.cpp
@@ -0,0 +1,39 @@
+#include "fake_probe.h"
+
+#include <gtest/gtest.h>
+
+namespace
+{
+
+using namespace std::chrono_literals;
+
+TEST(FakeProbe, ReturnsConfiguredResult)
+{
+ for(auto status : {ProbeStatus::GOOD, ProbeStatus::BAD, ProbeStatus::OTHER})
+ {
+ const ProbeResult configured{status, 123, 456};
+ FakeProbe fake(configured, 0ms);
+ ProbeInterface& probe = fake;
+ auto result = probe.probe();
+ ASSERT_TRUE(result);
+ EXPECT_EQ(result->status, status);
+ EXPECT_EQ(result->timestamp, configured.timestamp);
+ EXPECT_EQ(result->duration_microsecond, configured.duration_microsecond);
+ }
+}
+
+TEST(FakeProbe, DelaysWithoutChangingResult)
+{
+ const ProbeResult configured{ProbeStatus::BAD, 123, 456};
+ FakeProbe probe(configured, 20ms);
+ const auto start = std::chrono::steady_clock::now();
+ auto result = probe.probe();
+ const auto elapsed = std::chrono::steady_clock::now() - start;
+ ASSERT_TRUE(result);
+ EXPECT_EQ(result->status, configured.status);
+ EXPECT_GE(elapsed, 20ms);
+ EXPECT_EQ(result->timestamp, configured.timestamp);
+ EXPECT_EQ(result->duration_microsecond, configured.duration_microsecond);
+}
+
+}
diff --git a/tests/probe_test.cpp b/tests/probe_test.cpp
new file mode 100644
index 0000000..aa7ea34
--- /dev/null
+++ b/tests/probe_test.cpp
@@ -0,0 +1,414 @@
+#include "probe.h"
+
+#include <array>
+#include <cerrno>
+#include <cstring>
+#include <future>
+#include <stop_token>
+#include <thread>
+
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <poll.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <gtest/gtest.h>
+
+namespace
+{
+
+using namespace std::chrono_literals;
+
+bool waitReadable(int fd, std::stop_token stop)
+{
+ while(!stop.stop_requested())
+ {
+ pollfd descriptor{fd, POLLIN, 0};
+ const int ready = poll(&descriptor, 1, 20);
+ if(ready > 0)
+ {
+ return true;
+ }
+ if(ready < 0 && errno != EINTR)
+ {
+ return false;
+ }
+ }
+ return false;
+}
+
+class ProbeTest : public testing::Test
+{
+protected:
+ void TearDown() override
+ {
+ worker.request_stop();
+ if(worker.joinable())
+ {
+ worker.join();
+ }
+ if(server >= 0)
+ {
+ close(server);
+ }
+ }
+
+ bool bindServer(int type, int family = AF_INET, bool listening = false)
+ {
+ server = socket(family, type | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
+ if(server < 0)
+ {
+ return false;
+ }
+ sockaddr_storage address{};
+ socklen_t size = 0;
+ if(family == AF_INET)
+ {
+ auto& ipv4 = reinterpret_cast<sockaddr_in&>(address);
+ ipv4.sin_family = AF_INET;
+ ipv4.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+ size = sizeof(ipv4);
+ }
+ else
+ {
+ auto& ipv6 = reinterpret_cast<sockaddr_in6&>(address);
+ ipv6.sin6_family = AF_INET6;
+ ipv6.sin6_addr = in6addr_loopback;
+ size = sizeof(ipv6);
+ }
+ if(bind(server, reinterpret_cast<sockaddr*>(&address), size) < 0 ||
+ getsockname(server, reinterpret_cast<sockaddr*>(&address),
+ &size) < 0)
+ {
+ return false;
+ }
+ port = family == AF_INET
+ ? ntohs(reinterpret_cast<sockaddr_in&>(address).sin_port)
+ : ntohs(reinterpret_cast<sockaddr_in6&>(address).sin6_port);
+ return !listening || listen(server, 8) == 0;
+ }
+
+ static void serveHttp(std::stop_token stop, ProbeTest* self,
+ int status, bool silent)
+ {
+ if(!waitReadable(self->server, stop))
+ {
+ return;
+ }
+ const int client = accept4(self->server, nullptr, nullptr,
+ SOCK_NONBLOCK | SOCK_CLOEXEC);
+ if(client < 0)
+ {
+ return;
+ }
+ std::string request;
+ while(request.find("\r\n\r\n") == std::string::npos &&
+ request.size() < 8192 && waitReadable(client, stop))
+ {
+ std::array<char, 1024> buffer;
+ const auto count = recv(client, buffer.data(), buffer.size(), 0);
+ if(count <= 0)
+ {
+ break;
+ }
+ request.append(buffer.data(), count);
+ }
+ if(!silent)
+ {
+ const auto response = "HTTP/1.1 " + std::to_string(status) +
+ " Test\r\nContent-Length: 2\r\nConnection: close\r\n"
+ "Location: /redirect\r\n\r\nok";
+ send(client, response.data(), response.size(), MSG_NOSIGNAL);
+ }
+ else
+ {
+ // Wait for the probe to time out and close its connection.
+ while(waitReadable(client, stop))
+ {
+ char buffer;
+ if(recv(client, &buffer, 1, 0) <= 0)
+ {
+ break;
+ }
+ }
+ }
+ close(client);
+ }
+
+ static void serveUdp(std::stop_token stop, ProbeTest* self,
+ bool empty_reply)
+ {
+ if(!waitReadable(self->server, stop))
+ {
+ return;
+ }
+ std::array<char, 1024> buffer;
+ sockaddr_storage source{};
+ socklen_t size = sizeof(source);
+ const auto count = recvfrom(self->server, buffer.data(), buffer.size(),
+ 0, reinterpret_cast<sockaddr*>(&source), &size);
+ if(count >= 0)
+ {
+ self->received_payload.assign(buffer.data(), count);
+ const std::string reply = empty_reply ? "" : "reply";
+ sendto(self->server, reply.data(), reply.size(), MSG_NOSIGNAL,
+ reinterpret_cast<sockaddr*>(&source), size);
+ }
+ }
+
+ int server = -1;
+ std::uint16_t port = 0;
+ std::string received_payload;
+ std::jthread worker;
+};
+
+TEST(Probe, InvalidConfiguration)
+{
+ EXPECT_FALSE(createProbe(TcpEndpoint{"", 80}));
+ EXPECT_FALSE(createProbe(TcpEndpoint{"localhost", 0}));
+ EXPECT_FALSE(createProbe(UdpEndpoint{"localhost", 0, ""}));
+ EXPECT_FALSE(createProbe(UdpEndpoint{"localhost", 80,
+ std::string(65508, 'x')}));
+ EXPECT_FALSE(createProbe(IcmpEndpoint{""}));
+ EXPECT_FALSE(createProbe(HttpEndpoint{"file:///etc/hosts"}));
+ EXPECT_FALSE(createProbe(HttpEndpoint{"http://"}));
+ EXPECT_FALSE(createProbe(HttpEndpoint{"http://localhost"}, 0s));
+ EXPECT_FALSE(createProbe(TcpEndpoint{"localhost", 80}, -1s));
+ EXPECT_FALSE(createProbe(TcpEndpoint{
+ "localhost", 80}, std::chrono::seconds::max()));
+ EXPECT_FALSE(createProbe(UdpEndpoint{"localhost", 80, ""}, 0s));
+ EXPECT_FALSE(createProbe(IcmpEndpoint{"localhost"}, 0s));
+ EXPECT_FALSE(createProbe(TcpEndpoint{
+ std::string("host\0suffix", 11), 80}));
+}
+
+TEST(Probe, FactoryCreatesConcreteTypes)
+{
+ auto http = createProbe(HttpEndpoint{"https://example.invalid"});
+ ASSERT_TRUE(http);
+ EXPECT_NE(dynamic_cast<HttpProbe*>(http->get()), nullptr);
+ auto tcp = createProbe(TcpEndpoint{"localhost", 80});
+ ASSERT_TRUE(tcp);
+ EXPECT_NE(dynamic_cast<TcpProbe*>(tcp->get()), nullptr);
+ auto udp = createProbe(UdpEndpoint{"localhost", 53, ""});
+ ASSERT_TRUE(udp);
+ EXPECT_NE(dynamic_cast<UdpProbe*>(udp->get()), nullptr);
+ auto icmp = createProbe(IcmpEndpoint{"localhost"});
+ ASSERT_TRUE(icmp);
+ EXPECT_NE(dynamic_cast<IcmpProbe*>(icmp->get()), nullptr);
+}
+
+TEST_F(ProbeTest, OwnsConfiguration)
+{
+ ASSERT_TRUE(bindServer(SOCK_STREAM, AF_INET, true));
+ mw::E<std::unique_ptr<ProbeInterface>> probe;
+ {
+ TcpEndpoint config{"127.0.0.1", port};
+ probe = createProbe(config, 1s);
+ ASSERT_TRUE(probe) << probe.error().msg();
+ config.host.clear();
+ config.port = 0;
+ }
+ auto result = (*probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::GOOD);
+}
+
+mw::E<ProbeResult> runOwnedProbe(std::unique_ptr<ProbeInterface> probe)
+{
+ return probe->probe();
+}
+
+TEST_F(ProbeTest, TaskOwnsProbe)
+{
+ ASSERT_TRUE(bindServer(SOCK_STREAM, AF_INET, true));
+ auto probe = createProbe(TcpEndpoint{"127.0.0.1", port}, 1s);
+ ASSERT_TRUE(probe) << probe.error().msg();
+ auto task = std::async(std::launch::async, runOwnedProbe,
+ std::move(*probe));
+ EXPECT_EQ(probe->get(), nullptr);
+ auto result = task.get();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::GOOD);
+}
+
+TEST_F(ProbeTest, TcpConnectAndTiming)
+{
+ ASSERT_TRUE(bindServer(SOCK_STREAM, AF_INET, true));
+ const auto before = std::chrono::duration_cast<std::chrono::seconds>(
+ std::chrono::system_clock::now().time_since_epoch()).count();
+ auto result_probe = createProbe(TcpEndpoint{"127.0.0.1", port}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::GOOD);
+ EXPECT_GE(result->timestamp, before);
+ EXPECT_GE(result->duration_microsecond, 0);
+}
+
+TEST_F(ProbeTest, TcpIpv6)
+{
+ ASSERT_TRUE(bindServer(SOCK_STREAM, AF_INET6, true));
+ auto result_probe = createProbe(TcpEndpoint{"::1", port}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::GOOD);
+}
+
+TEST_F(ProbeTest, HostnameResolution)
+{
+ ASSERT_TRUE(bindServer(SOCK_STREAM, AF_INET, true));
+ auto result_probe = createProbe(TcpEndpoint{"localhost", port}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::GOOD);
+}
+
+TEST_F(ProbeTest, RefusedTcpAndHttp)
+{
+ ASSERT_TRUE(bindServer(SOCK_STREAM));
+ auto tcp_probe = createProbe(TcpEndpoint{"127.0.0.1", port}, 1s);
+ ASSERT_TRUE(tcp_probe) << tcp_probe.error().msg();
+ auto tcp = (*tcp_probe)->probe();
+ ASSERT_TRUE(tcp) << tcp.error().msg();
+ EXPECT_EQ(tcp->status, ProbeStatus::BAD);
+ auto http_probe = createProbe(HttpEndpoint{
+ "http://127.0.0.1:" + std::to_string(port)}, 1s);
+ ASSERT_TRUE(http_probe) << http_probe.error().msg();
+ auto http = (*http_probe)->probe();
+ ASSERT_TRUE(http) << http.error().msg();
+ EXPECT_EQ(http->status, ProbeStatus::BAD);
+}
+
+TEST_F(ProbeTest, HttpStatusCodes)
+{
+ ASSERT_TRUE(bindServer(SOCK_STREAM, AF_INET, true));
+ for(int status : {200, 204, 299, 301, 404, 503})
+ {
+ worker = std::jthread(&ProbeTest::serveHttp, this, status, false);
+ auto result_probe = createProbe(HttpEndpoint{
+ "http://127.0.0.1:" + std::to_string(port)}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, status >= 200 && status < 300
+ ? ProbeStatus::GOOD : ProbeStatus::BAD) << status;
+ worker.join();
+ }
+}
+
+TEST_F(ProbeTest, HttpTimeout)
+{
+ ASSERT_TRUE(bindServer(SOCK_STREAM, AF_INET, true));
+ worker = std::jthread(&ProbeTest::serveHttp, this, 200, true);
+ auto result_probe = createProbe(HttpEndpoint{
+ "http://127.0.0.1:" + std::to_string(port)}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::BAD);
+ EXPECT_GE(result->duration_microsecond, 900000);
+ EXPECT_LT(result->duration_microsecond, 3000000);
+}
+
+TEST_F(ProbeTest, UdpReplyAndBinaryPayload)
+{
+ ASSERT_TRUE(bindServer(SOCK_DGRAM));
+ worker = std::jthread(&ProbeTest::serveUdp, this, false);
+ const std::string payload("hello\0world", 11);
+ auto result_probe = createProbe(
+ UdpEndpoint{"127.0.0.1", port, payload}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::GOOD);
+ worker.join();
+ EXPECT_EQ(received_payload, payload);
+}
+
+TEST_F(ProbeTest, UdpEmptyReplyAndIpv6)
+{
+ ASSERT_TRUE(bindServer(SOCK_DGRAM, AF_INET6));
+ worker = std::jthread(&ProbeTest::serveUdp, this, true);
+ auto result_probe = createProbe(UdpEndpoint{"::1", port, ""}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::GOOD);
+ worker.join();
+ EXPECT_TRUE(received_payload.empty());
+}
+
+TEST_F(ProbeTest, UdpSilenceIsOther)
+{
+ ASSERT_TRUE(bindServer(SOCK_DGRAM));
+ auto result_probe = createProbe(
+ UdpEndpoint{"127.0.0.1", port, "probe"}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::OTHER);
+ EXPECT_GE(result->duration_microsecond, 900000);
+ EXPECT_LT(result->duration_microsecond, 3000000);
+}
+
+TEST_F(ProbeTest, UdpRefused)
+{
+ ASSERT_TRUE(bindServer(SOCK_DGRAM));
+ close(server);
+ server = -1;
+ auto result_probe = createProbe(
+ UdpEndpoint{"127.0.0.1", port, "probe"}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::BAD);
+}
+
+TEST(Probe, DnsFailureIsBad)
+{
+ auto result_probe = createProbe(
+ TcpEndpoint{"status-tracker.invalid", 80}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::BAD);
+ EXPECT_LT(result->duration_microsecond, 3000000);
+}
+
+TEST(Probe, IcmpIpv4)
+{
+ const int allowed = socket(AF_INET, SOCK_DGRAM, IPPROTO_ICMP);
+ auto result_probe = createProbe(IcmpEndpoint{"127.0.0.1"}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ if(allowed < 0)
+ {
+ EXPECT_FALSE(result);
+ GTEST_SKIP() << "ICMP ping sockets unavailable for this process";
+ }
+ close(allowed);
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::GOOD);
+}
+
+TEST(Probe, IcmpIpv6)
+{
+ const int allowed = socket(AF_INET6, SOCK_DGRAM, IPPROTO_ICMPV6);
+ auto result_probe = createProbe(IcmpEndpoint{"::1"}, 1s);
+ ASSERT_TRUE(result_probe) << result_probe.error().msg();
+ auto result = (*result_probe)->probe();
+ if(allowed < 0)
+ {
+ EXPECT_FALSE(result);
+ GTEST_SKIP() << "ICMPv6 ping sockets unavailable for this process";
+ }
+ close(allowed);
+ ASSERT_TRUE(result) << result.error().msg();
+ EXPECT_EQ(result->status, ProbeStatus::GOOD);
+}
+
+}
diff --git a/tests/scheduler_test.cpp b/tests/scheduler_test.cpp
new file mode 100644
index 0000000..1e81ef3
--- /dev/null
+++ b/tests/scheduler_test.cpp
@@ -0,0 +1,463 @@
+#include "scheduler.h"
+#include "data_source_sqlite.h"
+#include "fake_probe.h"
+
+#include <atomic>
+#include <future>
+#include <stdexcept>
+
+#include <gtest/gtest.h>
+
+namespace
+{
+
+using namespace std::chrono_literals;
+
+TEST(Scheduler, NextDueTimePreservesCadence)
+{
+ using Time = Scheduler::Clock::time_point;
+ auto next = Scheduler::nextDueTime(Time{10s}, 5s, Time{10s});
+ ASSERT_TRUE(next);
+ EXPECT_EQ(*next, Time{15s});
+ auto late = Scheduler::nextDueTime(Time{10s}, 5s, Time{27s});
+ ASSERT_TRUE(late);
+ EXPECT_EQ(*late, Time{30s});
+ auto exact = Scheduler::nextDueTime(Time{10s}, 5s, Time{30s});
+ ASSERT_TRUE(exact);
+ EXPECT_EQ(*exact, Time{35s});
+}
+
+TEST(Scheduler, NextDueTimeRejectsInvalidSchedule)
+{
+ using Time = Scheduler::Clock::time_point;
+ EXPECT_FALSE(Scheduler::nextDueTime(Time{}, 0s, Time{}));
+ EXPECT_FALSE(Scheduler::nextDueTime(Time{}, -1s, Time{}));
+ EXPECT_FALSE(Scheduler::nextDueTime(Time{-1s}, 1s, Time{}));
+ EXPECT_FALSE(Scheduler::nextDueTime(Time{1s}, 1s, Time{}));
+}
+
+TEST(Scheduler, NextDueTimeDetectsOverflow)
+{
+ using Time = Scheduler::Clock::time_point;
+ EXPECT_FALSE(Scheduler::nextDueTime(Time::max(), 1s, Time::max()));
+}
+
+// Keep SQLite persistence real while observing worker completion in tests.
+class WatchingSource final : public DataSourceInterface
+{
+public:
+ enum class SaveMode { SUCCESS, ERROR, EXCEPTION };
+
+ const Scheduler* scheduler = nullptr;
+ SaveMode mode = SaveMode::SUCCESS;
+ std::atomic<bool> flight_during_save{true};
+
+ mw::E<void> save(const StatusRecord& record) override
+ {
+ if(scheduler)
+ {
+ const auto* service = scheduler->findService(record.service_id);
+ if(!service || !service->in_flight.load())
+ {
+ flight_during_save = false;
+ }
+ }
+ if(mode == SaveMode::ERROR)
+ {
+ return std::unexpected(mw::runtimeError("Save failed"));
+ }
+ if(mode == SaveMode::EXCEPTION)
+ {
+ throw std::runtime_error("Unexpected save exception");
+ }
+ auto result = database->save(record);
+ if(result)
+ {
+ {
+ std::lock_guard lock(mutex);
+ ++counts[record.service_id];
+ }
+ changed.notify_all();
+ }
+ return result;
+ }
+
+ mw::E<std::optional<StatusRecord>> latest(
+ const std::string& service_id) override
+ {
+ return database->latest(service_id);
+ }
+
+ mw::E<std::vector<StatusRecord>> history(
+ const std::string& service_id, std::int64_t start,
+ std::int64_t end) override
+ {
+ return database->history(service_id, start, end);
+ }
+
+ bool waitForRecords(const std::string& service_id, std::size_t count,
+ std::chrono::milliseconds timeout = 1s)
+ {
+ std::unique_lock lock(mutex);
+ const auto deadline = std::chrono::steady_clock::now() + timeout;
+ while(counts[service_id] < count)
+ {
+ if(changed.wait_until(lock, deadline) == std::cv_status::timeout)
+ {
+ return counts[service_id] >= count;
+ }
+ }
+ return true;
+ }
+
+private:
+ std::unique_ptr<DataSourceSqlite> database =
+ DataSourceSqlite::open(":memory:").value();
+ std::mutex mutex;
+ std::condition_variable changed;
+ std::unordered_map<std::string, std::size_t> counts;
+};
+
+ServiceConfig serviceConfig(const std::string& id,
+ Scheduler::Clock::duration interval = 1h)
+{
+ return {id, "Display " + id, "Description", TcpEndpoint{"localhost", 80},
+ 1s, interval, "https://example.invalid"};
+}
+
+Configuration oneService()
+{
+ return {1, ":memory:", {{"Group", {serviceConfig("service")}}}};
+}
+
+mw::E<std::unique_ptr<ProbeInterface>> makeFake(
+ [[maybe_unused]] const Service& service)
+{
+ return std::make_unique<FakeProbe>(
+ ProbeResult{ProbeStatus::BAD, 123, 456}, 0ms);
+}
+
+// Cancel before the async run's future is destroyed after a failed assertion.
+struct StopOnExit
+{
+ std::stop_source& source;
+
+ ~StopOnExit()
+ {
+ source.request_stop();
+ }
+};
+
+TEST(Scheduler, BuildsServicesAcrossGroupsAndPersistsResults)
+{
+ WatchingSource source;
+ Configuration configuration{2, ":memory:", {
+ {"First", {serviceConfig("first")}},
+ {"Second", {serviceConfig("second")}}
+ }};
+ auto created = Scheduler::create(configuration, source, makeFake);
+ ASSERT_TRUE(created) << created.error().msg();
+ auto scheduler = std::move(*created);
+ source.scheduler = scheduler.get();
+ ASSERT_NE(scheduler->findService("first"), nullptr);
+ ASSERT_NE(scheduler->findService("second"), nullptr);
+ EXPECT_EQ(scheduler->findService("missing"), nullptr);
+ EXPECT_EQ(scheduler->findService("first")->config.name, "Display first");
+ configuration.groups.clear();
+ EXPECT_EQ(scheduler->findService("first")->config.id, "first");
+
+ std::stop_source stop;
+ auto running = std::async(std::launch::async, &Scheduler::run,
+ scheduler.get(), stop.get_token());
+ StopOnExit cleanup{stop};
+ ASSERT_TRUE(source.waitForRecords("first", 1));
+ ASSERT_TRUE(source.waitForRecords("second", 1));
+ stop.request_stop();
+ ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
+ EXPECT_TRUE(running.get());
+ for(const auto& id : {"first", "second"})
+ {
+ auto record = source.latest(id);
+ ASSERT_TRUE(record);
+ ASSERT_TRUE(*record);
+ EXPECT_EQ((**record), (StatusRecord{id, 123, 456, ProbeStatus::BAD}));
+ EXPECT_FALSE(scheduler->findService(id)->in_flight.load());
+ }
+ EXPECT_TRUE(source.flight_during_save.load());
+ EXPECT_FALSE(scheduler->run(stop.get_token()));
+}
+
+TEST(Scheduler, CancellationWakesEmptyQueue)
+{
+ WatchingSource source;
+ auto scheduler = Scheduler::create(
+ Configuration{1, ":memory:", {}}, source);
+ ASSERT_TRUE(scheduler);
+ std::stop_source stop;
+ auto running = std::async(std::launch::async, &Scheduler::run,
+ scheduler->get(), stop.get_token());
+ StopOnExit cleanup{stop};
+ EXPECT_EQ(running.wait_for(20ms), std::future_status::timeout);
+ stop.request_stop();
+ ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
+ EXPECT_TRUE(running.get());
+}
+
+mw::E<std::unique_ptr<ProbeInterface>> factoryError(
+ [[maybe_unused]] const Service& service)
+{
+ return std::unexpected(mw::runtimeError("Factory failed"));
+}
+
+TEST(Scheduler, CancellationBeforeStart)
+{
+ WatchingSource source;
+ auto scheduler = Scheduler::create(oneService(), source, factoryError);
+ ASSERT_TRUE(scheduler);
+ std::stop_source stop;
+ stop.request_stop();
+ EXPECT_TRUE((*scheduler)->run(stop.get_token()));
+ EXPECT_FALSE((*scheduler)->findService("service")->in_flight.load());
+ EXPECT_FALSE(source.latest("service").value());
+}
+
+TEST(Scheduler, CancellationWakesFutureDueTime)
+{
+ WatchingSource source;
+ auto scheduler = Scheduler::create(oneService(), source, makeFake);
+ ASSERT_TRUE(scheduler);
+ std::stop_source stop;
+ auto running = std::async(std::launch::async, &Scheduler::run,
+ scheduler->get(), stop.get_token());
+ StopOnExit cleanup{stop};
+ ASSERT_TRUE(source.waitForRecords("service", 1));
+ EXPECT_EQ(running.wait_for(20ms), std::future_status::timeout);
+ stop.request_stop();
+ ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
+ EXPECT_TRUE(running.get());
+}
+
+class FailureProbe final : public ProbeInterface
+{
+public:
+ explicit FailureProbe(bool throw_exception)
+ : throw_exception(throw_exception)
+ {}
+
+ mw::E<ProbeResult> probe() override
+ {
+ if(throw_exception)
+ {
+ throw std::runtime_error("Unexpected probe exception");
+ }
+ return std::unexpected(mw::runtimeError("Probe failed"));
+ }
+
+private:
+ bool throw_exception;
+};
+
+mw::E<std::unique_ptr<ProbeInterface>> failedProbe(
+ [[maybe_unused]] const Service& service)
+{
+ return std::make_unique<FailureProbe>(false);
+}
+
+mw::E<std::unique_ptr<ProbeInterface>> throwingProbe(
+ [[maybe_unused]] const Service& service)
+{
+ return std::make_unique<FailureProbe>(true);
+}
+
+mw::E<std::unique_ptr<ProbeInterface>> nullProbe(
+ [[maybe_unused]] const Service& service)
+{
+ return std::unique_ptr<ProbeInterface>{};
+}
+
+mw::E<std::unique_ptr<ProbeInterface>> throwingFactory(
+ [[maybe_unused]] const Service& service)
+{
+ throw std::runtime_error("Unexpected factory exception");
+}
+
+TEST(Scheduler, FailuresClearServiceFlightState)
+{
+ for(auto factory : {factoryError, failedProbe, throwingProbe,
+ nullProbe, throwingFactory})
+ {
+ WatchingSource source;
+ auto scheduler = Scheduler::create(oneService(), source, factory);
+ ASSERT_TRUE(scheduler);
+ std::stop_source stop;
+ auto running = std::async(std::launch::async, &Scheduler::run,
+ scheduler->get(), stop.get_token());
+ StopOnExit cleanup{stop};
+ ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
+ auto result = running.get();
+ ASSERT_FALSE(result);
+ EXPECT_NE(result.error().msg().find("service:"), std::string::npos);
+ EXPECT_FALSE((*scheduler)->findService("service")->in_flight.load());
+ EXPECT_FALSE(source.latest("service").value());
+ }
+}
+
+TEST(Scheduler, PersistenceFailuresClearServiceFlightState)
+{
+ for(auto mode : {WatchingSource::SaveMode::ERROR,
+ WatchingSource::SaveMode::EXCEPTION})
+ {
+ WatchingSource source;
+ source.mode = mode;
+ auto scheduler = Scheduler::create(oneService(), source, makeFake);
+ ASSERT_TRUE(scheduler);
+ source.scheduler = scheduler->get();
+ std::stop_source stop;
+ auto running = std::async(std::launch::async, &Scheduler::run,
+ scheduler->get(), stop.get_token());
+ StopOnExit cleanup{stop};
+ ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
+ auto result = running.get();
+ ASSERT_FALSE(result);
+ EXPECT_NE(result.error().msg().find("service:"), std::string::npos);
+ EXPECT_FALSE((*scheduler)->findService("service")->in_flight.load());
+ EXPECT_TRUE(source.flight_during_save.load());
+ }
+}
+
+class BlockingProbe final : public ProbeInterface
+{
+public:
+ BlockingProbe(std::shared_future<void> release, std::promise<void>& started)
+ : release(std::move(release)), started(started)
+ {}
+
+ mw::E<ProbeResult> probe() override
+ {
+ started.set_value();
+ release.wait();
+ return ProbeResult{ProbeStatus::GOOD, 123, 456};
+ }
+
+private:
+ std::shared_future<void> release;
+ std::promise<void>& started;
+};
+
+struct BlockingFactory
+{
+ std::shared_future<void> release;
+ std::promise<void>& started;
+ std::atomic<int>& calls;
+
+ mw::E<std::unique_ptr<ProbeInterface>> operator()(
+ const Service& service) const
+ {
+ if(service.config.id == "a_slow")
+ {
+ ++calls;
+ return std::make_unique<BlockingProbe>(release, started);
+ }
+ return makeFake(service);
+ }
+};
+
+TEST(Scheduler, SkipsInFlightAndFinishesAcceptedWorkOnStop)
+{
+ WatchingSource source;
+ std::atomic<int> calls{0};
+ std::promise<void> started;
+ auto ready = started.get_future();
+ std::stop_source stop;
+ auto release = std::make_unique<std::promise<void>>();
+ Configuration configuration{2, ":memory:", {{"Group", {
+ serviceConfig("a_slow", 1ms), serviceConfig("ticker", 1ms)
+ }}}};
+ auto scheduler = Scheduler::create(configuration, source, BlockingFactory{
+ release->get_future().share(), started, calls});
+ ASSERT_TRUE(scheduler);
+ source.scheduler = scheduler->get();
+ auto running = std::async(std::launch::async, &Scheduler::run,
+ scheduler->get(), stop.get_token());
+ // Unblock the worker before destroying running on assertion failure.
+ auto release_on_exit = std::move(release);
+ StopOnExit cleanup{stop};
+ ASSERT_EQ(ready.wait_for(1s), std::future_status::ready);
+ ASSERT_TRUE(source.waitForRecords("ticker", 4));
+ EXPECT_EQ(calls.load(), 1);
+ EXPECT_TRUE((*scheduler)->findService("a_slow")->in_flight.load());
+ stop.request_stop();
+ EXPECT_EQ(running.wait_for(20ms), std::future_status::timeout);
+ release_on_exit->set_value();
+ ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
+ EXPECT_TRUE(running.get());
+ EXPECT_FALSE((*scheduler)->findService("a_slow")->in_flight.load());
+ EXPECT_FALSE((*scheduler)->findService("ticker")->in_flight.load());
+ EXPECT_TRUE(source.flight_during_save.load());
+ EXPECT_TRUE(source.latest("a_slow").value());
+}
+
+TEST(Scheduler, SkipsWhenFullAndRetriesAtNextInterval)
+{
+ WatchingSource source;
+ std::atomic<int> calls{0};
+ std::promise<void> started;
+ auto ready = started.get_future();
+ std::stop_source stop;
+ auto release = std::make_unique<std::promise<void>>();
+ Configuration configuration{1, ":memory:", {{"Group", {
+ serviceConfig("a_slow"), serviceConfig("retry", 1ms)
+ }}}};
+ auto scheduler = Scheduler::create(configuration, source, BlockingFactory{
+ release->get_future().share(), started, calls});
+ ASSERT_TRUE(scheduler);
+ auto running = std::async(std::launch::async, &Scheduler::run,
+ scheduler->get(), stop.get_token());
+ auto release_on_exit = std::move(release);
+ StopOnExit cleanup{stop};
+ ASSERT_EQ(ready.wait_for(1s), std::future_status::ready);
+ EXPECT_FALSE(source.waitForRecords("retry", 1, 20ms));
+ release_on_exit->set_value();
+ ASSERT_TRUE(source.waitForRecords("retry", 1));
+ stop.request_stop();
+ ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
+ EXPECT_TRUE(running.get());
+ EXPECT_FALSE((*scheduler)->findService("retry")->in_flight.load());
+ EXPECT_EQ(calls.load(), 1);
+}
+
+TEST(Scheduler, InvalidConfiguration)
+{
+ WatchingSource source;
+ auto configuration = oneService();
+ configuration.worker_count = 0;
+ EXPECT_FALSE(Scheduler::create(configuration, source));
+ configuration = oneService();
+ EXPECT_FALSE(Scheduler::create(configuration, source, {}));
+ configuration.groups.front().services.front().id.clear();
+ EXPECT_FALSE(Scheduler::create(configuration, source));
+ configuration = oneService();
+ configuration.groups.front().services.front().interval = 0s;
+ EXPECT_FALSE(Scheduler::create(configuration, source));
+ configuration.groups.front().services.front().interval = -1s;
+ EXPECT_FALSE(Scheduler::create(configuration, source));
+ configuration.groups.front().services.front().interval =
+ Scheduler::Clock::duration::max();
+ EXPECT_FALSE(Scheduler::create(configuration, source));
+ configuration = oneService();
+ configuration.groups.front().services.front().timeout = 0s;
+ EXPECT_FALSE(Scheduler::create(configuration, source));
+ configuration = oneService();
+ configuration.groups.front().services.front().endpoint =
+ TcpEndpoint{"localhost", 0};
+ EXPECT_FALSE(Scheduler::create(configuration, source));
+ configuration = oneService();
+ configuration.groups.push_back(
+ {"Another group", {serviceConfig("service")}});
+ auto duplicate = Scheduler::create(configuration, source);
+ ASSERT_FALSE(duplicate);
+ EXPECT_NE(duplicate.error().msg().find("Duplicate service ID"),
+ std::string::npos);
+}
+
+}
diff --git a/tests/service_test.cpp b/tests/service_test.cpp
new file mode 100644
index 0000000..2eb4dd9
--- /dev/null
+++ b/tests/service_test.cpp
@@ -0,0 +1,82 @@
+#include "service.h"
+#include "configuration.h"
+
+#include <gtest/gtest.h>
+
+namespace
+{
+
+using namespace std::chrono_literals;
+
+TEST(Service, CreatesFreshProtocolProbes)
+{
+ ServiceConfig config{"blog", "Blog", "Personal blog",
+ HttpEndpoint{"https://example.invalid"}, 5s, 1min,
+ std::nullopt};
+ Configuration configuration{2, ":memory:", {{"Personal", {config}}}};
+ Service service{configuration.groups.front().services.front()};
+ auto first = service.createProbe();
+ ASSERT_TRUE(first) << first.error().msg();
+ EXPECT_NE(dynamic_cast<HttpProbe*>(first->get()), nullptr);
+ auto second = service.createProbe();
+ ASSERT_TRUE(second) << second.error().msg();
+ EXPECT_NE(first->get(), second->get());
+ EXPECT_FALSE(service.in_flight.load());
+
+ service.config.endpoint = TcpEndpoint{"localhost", 80};
+ auto tcp = service.createProbe();
+ ASSERT_TRUE(tcp) << tcp.error().msg();
+ EXPECT_NE(dynamic_cast<TcpProbe*>(tcp->get()), nullptr);
+
+ service.config.endpoint = UdpEndpoint{"localhost", 53, ""};
+ auto udp = service.createProbe();
+ ASSERT_TRUE(udp) << udp.error().msg();
+ EXPECT_NE(dynamic_cast<UdpProbe*>(udp->get()), nullptr);
+
+ service.config.endpoint = IcmpEndpoint{"localhost"};
+ auto icmp = service.createProbe();
+ ASSERT_TRUE(icmp) << icmp.error().msg();
+ EXPECT_NE(dynamic_cast<IcmpProbe*>(icmp->get()), nullptr);
+ EXPECT_TRUE(std::holds_alternative<HttpEndpoint>(
+ configuration.groups.front().services.front().endpoint));
+}
+
+TEST(Service, HttpUrlFallback)
+{
+ Service service{{"blog", "Blog", "", HttpEndpoint{}, 5s, 1min,
+ "https://example.invalid"}};
+ EXPECT_TRUE(service.createProbe());
+ EXPECT_TRUE(std::get<HttpEndpoint>(service.config.endpoint).url.empty());
+
+ service.config.url.reset();
+ EXPECT_FALSE(service.createProbe());
+
+ service.config.url = "invalid URL";
+ EXPECT_FALSE(service.createProbe());
+
+ service.config.endpoint = HttpEndpoint{"https://probe.example.invalid"};
+ EXPECT_TRUE(service.createProbe());
+
+ service.config.endpoint = HttpEndpoint{"invalid endpoint URL"};
+ service.config.url = "https://example.invalid";
+ EXPECT_FALSE(service.createProbe());
+}
+
+TEST(Service, UsesServiceTimeoutAndPreservesFlightState)
+{
+ Service service{{"host", "Host", "", TcpEndpoint{"localhost", 80},
+ 0s, 1min, std::nullopt}};
+ EXPECT_FALSE(service.createProbe());
+ EXPECT_FALSE(service.in_flight.load());
+
+ service.config.timeout = 5s;
+ service.in_flight = true;
+ EXPECT_TRUE(service.createProbe());
+ EXPECT_TRUE(service.in_flight.load());
+
+ service.config.endpoint = TcpEndpoint{"localhost", 0};
+ EXPECT_FALSE(service.createProbe());
+ EXPECT_TRUE(service.in_flight.load());
+}
+
+}