Cleanup

.gitea/workflows/ci.yml

@@ -50,7 +50,7 @@ jobs:
         run: python3 modulus.py configure --target x86_64_gcc12_release --update
       - name: Check code style
         run: |
-          python3 -m pip install --no-cache-dir -r build/build_tools/docker/format-requirements.txt
+          python3 -m pip install --no-cache-dir -r build/build_tools/docker/format_requirements.txt
           python3 modulus.py format --check
       - name: Build native release
         run: python3 modulus.py build --target x86_64_gcc12_release

CMakeLists.txt

@@ -23,6 +23,8 @@ add_library(ligo
   src/service.cpp
   src/tcp.cpp
   src/udp.cpp
+  src/detail/dtls.cpp
+  src/detail/udp_protocol.cpp
 )
 target_include_directories(ligo PUBLIC
   $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
@@ -41,7 +43,7 @@ if(BUILD_TESTING)
 endif()
 
 install(TARGETS ligo
-  EXPORT ligo-targets
+  EXPORT ligo_targets
   ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}
   LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}
   RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
@@ -49,25 +51,25 @@ install(TARGETS ligo
 
 install(DIRECTORY include/ DESTINATION ${CMAKE_INSTALL_INCLUDEDIR})
 
-install(EXPORT ligo-targets
+install(EXPORT ligo_targets
   NAMESPACE ligo::
   DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/ligo
 )
 
 configure_package_config_file(
   ${CMAKE_CURRENT_SOURCE_DIR}/cmake/ligo_config.cmake.in
-  ${CMAKE_CURRENT_BINARY_DIR}/ligo-config.cmake
+  ${CMAKE_CURRENT_BINARY_DIR}/ligoConfig.cmake
   INSTALL_DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/ligo
 )
 
 write_basic_package_version_file(
-  ${CMAKE_CURRENT_BINARY_DIR}/ligo-config-version.cmake
+  ${CMAKE_CURRENT_BINARY_DIR}/ligoConfigVersion.cmake
   VERSION ${PROJECT_VERSION}
   COMPATIBILITY SameMajorVersion
 )
 
 install(FILES
-  ${CMAKE_CURRENT_BINARY_DIR}/ligo-config.cmake
-  ${CMAKE_CURRENT_BINARY_DIR}/ligo-config-version.cmake
+  ${CMAKE_CURRENT_BINARY_DIR}/ligoConfig.cmake
+  ${CMAKE_CURRENT_BINARY_DIR}/ligoConfigVersion.cmake
   DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/ligo
 )

README.md

@@ -4,6 +4,8 @@ Ligo is a C++23 networking and transport library. It provides HTTP JSON calls, f
 
 Ligo uses Modulus for shared build tooling, Boost.JSON, and OpenSSL integration. Product-specific routes and workflows belong in consuming applications.
 
+UDP reliability and socket lifecycle use separate private DTLS and wire-protocol adapters. Session-drain waits serialize timer registration and notifications on a waiter strand and retain one absolute shutdown deadline, including when notifications arrive before a wait is armed.
+
 ## Build
 
 Code style is managed by Modulus: run `./modulus.py format` or `./modulus.py format --check`. The generated Neovim integration uses the same pinned container formatter for C++ and Python, including existing format-on-save toggles. Configure first to prepare tooling; customize `tooling.editor.clang_format` and `tooling.format.exclude` in `modulus.json`, not generated editor files. See the shared Modulus README for the compact layout and no-alias policy.
@@ -43,3 +45,9 @@ python3 modulus.py package --all --replace
 ## License
 
 Ligo is released under the MIT License.
+
+### HTTP metadata and raw responses
+
+`request_context` includes lowercase HTTP `headers` and a `query` map. Set `http::server_options::parse_query = true` to route on the path and decode query parameters (`+` for spaces, percent escapes); malformed escapes and duplicate keys return a bad-request response. The default remains false, preserving existing exact-target routing. Request cookies retain semicolon separation when repeated.
+
+`service_response` accepts an HTTP `status` (default 200) and repeated response `headers`, for example `Set-Cookie`. Existing `json_response`, `text_response`, and aggregate initialization continue to work. `http::request` accepts additional headers and an optional raw body. `http::client::exchange()` returns status, lowercase headers, and an uninterpreted body, including non-2xx and 304 responses; `call()` retains the original JSON-envelope behavior. Repeated raw response headers are comma-joined in the map; callers needing multiple Set-Cookie values must not treat that combined value as a single cookie. TLS verification and body/time limits apply to both client paths.

cmake/ligo_config.cmake.in

@@ -3,4 +3,4 @@
 include(CMakeFindDependencyMacro)
 find_dependency(modulus CONFIG REQUIRED COMPONENTS boost_json openssl)
 
-include("${CMAKE_CURRENT_LIST_DIR}/ligo-targets.cmake")
+include("${CMAKE_CURRENT_LIST_DIR}/ligo_targets.cmake")

include/ligo/detail/lifecycle.hpp

@@ -9,6 +9,8 @@
 #include <vector>
 
 #include <boost/asio/as_tuple.hpp>
+#include <boost/asio/co_spawn.hpp>
+#include <boost/asio/strand.hpp>
 #include <boost/asio/awaitable.hpp>
 #include <boost/asio/error.hpp>
 #include <boost/asio/post.hpp>
@@ -40,6 +42,7 @@ struct session_registry {
   auto cancel_all() -> void;
   auto empty() const -> std::bool_t;
   auto wait_empty(std::chrono::milliseconds timeout) -> boost::asio::awaitable<std::expected<void, modulus::error>>;
+  auto wait_registered(std::shared_ptr<boost::asio::steady_timer> timer, std::chrono::steady_clock::time_point deadline) -> boost::asio::awaitable<std::expected<void, modulus::error>>;
   auto notify_waiters() -> void;
   auto live_waiters_locked() -> std::vector<std::shared_ptr<boost::asio::steady_timer>>;
   mutable std::mutex mutex{};
@@ -74,7 +77,8 @@ auto ligo::detail::cancel_session_stream(boost::asio::ssl::stream<NEXT_LAYER> &s
 template <typename S>
 auto ligo::detail::session_registry<S>::cancel_waiters(std::vector<std::shared_ptr<boost::asio::steady_timer>> const &live) -> void {
   for (auto &waiter : live) {
-    waiter->cancel();
+    // Timer operations belong to the waiter strand, including registration.
+    boost::asio::post(waiter->get_executor(), [waiter] { waiter->cancel(); });
   }
 }
 
@@ -125,30 +129,28 @@ auto ligo::detail::session_registry<S>::wait_empty(std::chrono::milliseconds tim
   if (empty()) {
     co_return std::expected<void, modulus::error>{};
   }
-  auto notify_timer = std::make_shared<boost::asio::steady_timer>(executor);
-  auto deadline_timer = std::make_shared<boost::asio::steady_timer>(executor);
-  auto timed_out = std::make_shared<std::atomic<std::bool_t>>(false);
+  auto timer = std::make_shared<boost::asio::steady_timer>(boost::asio::make_strand(executor));
+  auto deadline = std::chrono::steady_clock::now() + timeout;
   {
     auto lock = std::scoped_lock{mutex};
-    waiters.push_back(notify_timer);
+    waiters.push_back(timer);
   }
-  deadline_timer->expires_after(timeout);
-  deadline_timer->async_wait([notify_timer, timed_out](boost::system::error_code const &error) {
-    if (!error) {
-      timed_out->store(true, std::memory_order_relaxed);
-      notify_timer->cancel();
-    }
-  });
+  co_return co_await boost::asio::co_spawn(timer->get_executor(), wait_registered(timer, deadline), boost::asio::use_awaitable);
+}
+
+template <typename S>
+auto ligo::detail::session_registry<S>::wait_registered(std::shared_ptr<boost::asio::steady_timer> timer, std::chrono::steady_clock::time_point deadline) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
+  // This entire coroutine runs on the timer strand. Notifications cannot run
+  // between checking session state and initiating the asynchronous wait.
   for (;;) {
     if (empty()) {
-      deadline_timer->cancel();
       co_return std::expected<void, modulus::error>{};
     }
-    if (timed_out->load(std::memory_order_relaxed)) {
+    if (std::chrono::steady_clock::now() >= deadline) {
       co_return ligo::detail::fail_void(ligo::detail::error_timeout, "timed out waiting for sessions to drain");
     }
-    notify_timer->expires_at(std::chrono::steady_clock::time_point::max());
-    auto [error] = co_await notify_timer->async_wait(boost::asio::as_tuple(boost::asio::use_awaitable));
+    timer->expires_at(deadline);
+    auto [error] = co_await timer->async_wait(boost::asio::as_tuple(boost::asio::use_awaitable));
     if (error && error != boost::asio::error::operation_aborted) {
       co_return ligo::detail::fail_void(ligo::detail::network_failure(error));
     }

include/ligo/http.hpp

@@ -4,6 +4,8 @@
 #include <chrono>
 #include <cstdint>
 #include <functional>
+#include <map>
+#include <optional>
 #include <string>
 
 #include <boost/asio/any_io_executor.hpp>
@@ -37,6 +39,14 @@ struct request {
   ligo::http::method method{ligo::http::method::POST};
   std::string target{"/"};
   boost::json::value payload{};
+  std::map<std::string, std::string> headers{};
+  std::optional<std::string> raw_body{};
+};
+
+struct response {
+  std::uint16_t status{};
+  std::map<std::string, std::string> headers{};
+  std::string body{};
 };
 
 struct client_options {
@@ -50,12 +60,14 @@ struct server_options {
   std::uint64_t max_body_size{16U * 1024U * 1024U};
   std::function<void(modulus::error)> on_error{};
   ligo::security_config security{};
+  bool parse_query{false};
 };
 
 class client {
 public:
   explicit client(ligo::http::endpoint endpoint, ligo::http::client_options options = {});
   auto call(ligo::http::request const &request) const -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>>;
+  auto exchange(ligo::http::request const &request) const -> boost::asio::awaitable<std::expected<ligo::http::response, modulus::error>>;
 private:
   ligo::http::endpoint endpoint_{};
   ligo::http::client_options options_{};

include/ligo/service.hpp

@@ -7,6 +7,8 @@
 #include <mutex>
 #include <string>
 #include <utility>
+#include <vector>
+#include <cstdint>
 
 #include <boost/asio/awaitable.hpp>
 #include <boost/json/value.hpp>
@@ -20,11 +22,15 @@ struct request_context {
   std::string peer{};
   std::string method{};
   std::string route{};
+  std::map<std::string, std::string> headers{};
+  std::map<std::string, std::string> query{};
 };
 
 struct service_response {
   std::string content_type{"application/json"};
   std::string body{};
+  std::uint16_t status{200};
+  std::vector<std::pair<std::string, std::string>> headers{};
 };
 
 class service_router {

modulus.py

@@ -9,7 +9,7 @@ import sys
 
 PACKAGE_REF = "modulus/latest"
 CONAN_VERSION = "2.26.2"
-FETCH_IMAGE = "modulus-fetch:dev"
+FETCH_IMAGE = "modulus_fetch:dev"
 COMMANDS = (
   "configure",
   "build",
@@ -19,6 +19,8 @@ COMMANDS = (
   "clangd",
   "format",
   "run",
+  "python",
+  "publish",
 )
 COMMON_DIR = pathlib.Path("build") / "build_common"
 TOOLS_DIR = pathlib.Path("build") / "build_tools"
@@ -26,7 +28,7 @@ RUNTIME_DIR = COMMON_DIR / "modulus"
 CONAN_HOME_DIR = COMMON_DIR / "conan_cache"
 DEFAULT_REMOTE_NAME = "khodo"
 DEFAULT_REMOTE_URL = "https://git.thyme-labs.com/api/packages/khodo/conan"
-DEFAULT_DOCKER_MARKER = "/opt/modulus-build-image"
+DEFAULT_DOCKER_MARKER = "/opt/modulus_build_image"
 
 def repo_root():
   root = pathlib.Path(__file__).resolve().parent
@@ -66,6 +68,9 @@ def runtime_ready(root):
     and (root / TOOLS_DIR / "format.py").is_file()
     and (root / RUNTIME_DIR / "modulus" / "formatting.py").is_file()
     and (root / RUNTIME_DIR / "modulus" / "format_layout.py").is_file()
+    and (root / RUNTIME_DIR / "modulus" / "python.py").is_file()
+    and (root / RUNTIME_DIR / "modulus" / "artifacts.py").is_file()
+    and (root / TOOLS_DIR / "python_runner.py").is_file()
   )
 
 def env_flag(name):
@@ -114,9 +119,9 @@ RUN apt-get update \\
     python3-venv \\
   && rm -rf /var/lib/apt/lists/*
 
-RUN python3 -m venv /opt/modulus-python
+RUN python3 -m venv /opt/modulus_python
 
-ENV PATH=/opt/modulus-python/bin:$PATH
+ENV PATH=/opt/modulus_python/bin:$PATH
 
 RUN python -m pip install --no-cache-dir conan=={CONAN_VERSION}
 

src/detail/dtls.cpp

@@ -0,0 +1,267 @@
+#include "dtls.hpp"
+
+#include <algorithm>
+#include <cstddef>
+#include <utility>
+#include <openssl/err.h>
+#include <openssl/ssl.h>
+#include <openssl/x509_vfy.h>
+#include <ligo/detail/errors.hpp>
+#include <ligo/detail/transport.hpp>
+#include <ligo/security.hpp>
+
+
+namespace ligo::detail {
+
+constexpr auto max_datagram_size = std::size_t{65507};
+constexpr auto max_dtls_plaintext_size = std::size_t{12000};
+
+auto dtls_failure(std::int32_t code, std::string message) -> modulus::error {
+  auto const error = ERR_get_error();
+  if (error != 0) {
+    message += ": ";
+    message += ERR_error_string(error, nullptr);
+  }
+  return ligo::detail::make_error(code, std::move(message));
+}
+
+auto dtls_failure(std::string message) -> modulus::error {
+  return dtls_failure(ligo::detail::error_unavailable, std::move(message));
+}
+
+auto validate_security(ligo::security_config const &config, std::bool_t server) -> std::expected<void, modulus::error> {
+  if (config.mode == ligo::security_mode::TLS) {
+    return ligo::detail::fail_void(ligo::unsupported_security("UDP", config.mode));
+  }
+  if (config.mode != ligo::security_mode::DTLS) {
+    return {};
+  }
+  if (config.tls_context) {
+    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS does not accept a TLS context"));
+  }
+  if (server && (config.certificate_chain_file.empty() || config.private_key_file.empty())) {
+    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS server requires certificate chain/private key files"));
+  }
+  if (!config.certificate_chain_file.empty() != !config.private_key_file.empty()) {
+    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS certificate chain and private key must both be configured"));
+  }
+  return {};
+}
+
+auto load_dtls_verify_paths(SSL_CTX *context, ligo::security_config const &config) -> std::expected<void, modulus::error> {
+  if (config.verify == ligo::verify_mode::NONE) {
+    return {};
+  }
+  if (!config.ca_file.empty()) {
+    if (SSL_CTX_load_verify_locations(context, config.ca_file.c_str(), nullptr) != 1) {
+      return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS CA file"));
+    }
+    return {};
+  }
+  if (SSL_CTX_set_default_verify_paths(context) != 1) {
+    return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS default verify paths"));
+  }
+  return {};
+}
+
+auto load_dtls_identity(SSL_CTX *context, ligo::security_config const &config) -> std::expected<void, modulus::error> {
+  if (config.certificate_chain_file.empty() && config.private_key_file.empty()) {
+    return {};
+  }
+  if (SSL_CTX_use_certificate_chain_file(context, config.certificate_chain_file.c_str()) != 1) {
+    return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS certificate chain"));
+  }
+  if (SSL_CTX_use_PrivateKey_file(context, config.private_key_file.c_str(), SSL_FILETYPE_PEM) != 1) {
+    return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS private key"));
+  }
+  if (SSL_CTX_check_private_key(context) != 1) {
+    return ligo::detail::fail_void(dtls_failure(ligo::detail::error_invalid_argument, "UDP DTLS private key does not match certificate"));
+  }
+  return {};
+}
+
+auto make_dtls_state(ligo::security_config const &config, std::bool_t server) -> std::expected<std::unique_ptr<dtls_state>, modulus::error> {
+  if (config.mode != ligo::security_mode::DTLS) {
+    return modulus::success(std::unique_ptr<dtls_state>{});
+  }
+  auto valid = validate_security(config, server);
+  if (!valid) {
+    return ligo::detail::fail<std::unique_ptr<dtls_state>>(valid.error());
+  }
+  auto context = std::unique_ptr<SSL_CTX, ssl_ctx_deleter>{SSL_CTX_new(server ? DTLS_server_method() : DTLS_client_method())};
+  if (!context) {
+    return ligo::detail::fail<std::unique_ptr<dtls_state>>(ligo::detail::error_unavailable, "failed to create UDP DTLS context");
+  }
+  SSL_CTX_set_options(context.get(), SSL_OP_NO_COMPRESSION | SSL_OP_NO_QUERY_MTU);
+  if (SSL_CTX_set_min_proto_version(context.get(), DTLS1_2_VERSION) != 1) {
+    return ligo::detail::fail<std::unique_ptr<dtls_state>>(dtls_failure("failed to configure UDP DTLS minimum protocol"));
+  }
+  auto const verify_peer = config.verify == ligo::verify_mode::PEER || (!server && config.verify == ligo::verify_mode::DEFAULTED);
+  auto const verify = verify_peer ? SSL_VERIFY_PEER | (server ? SSL_VERIFY_FAIL_IF_NO_PEER_CERT : 0) : SSL_VERIFY_NONE;
+  SSL_CTX_set_verify(context.get(), verify, nullptr);
+  auto verify_paths = load_dtls_verify_paths(context.get(), config);
+  if (!verify_paths) {
+    return ligo::detail::fail<std::unique_ptr<dtls_state>>(verify_paths.error());
+  }
+  auto identity = load_dtls_identity(context.get(), config);
+  if (!identity) {
+    return ligo::detail::fail<std::unique_ptr<dtls_state>>(identity.error());
+  }
+  auto ssl = std::unique_ptr<SSL, ssl_deleter>{SSL_new(context.get())};
+  if (!ssl) {
+    return ligo::detail::fail<std::unique_ptr<dtls_state>>(ligo::detail::error_unavailable, "failed to create UDP DTLS session");
+  }
+  if (!server && verify_peer && !config.peer_name.empty()) {
+    if (X509_VERIFY_PARAM_set1_host(SSL_get0_param(ssl.get()), config.peer_name.c_str(), 0) != 1) {
+      return ligo::detail::fail<std::unique_ptr<dtls_state>>(dtls_failure("failed to configure UDP DTLS peer identity"));
+    }
+  }
+  SSL_set_mtu(ssl.get(), 1200);
+  auto *ssl_bio = static_cast<BIO *>(nullptr);
+  auto *network_bio = static_cast<BIO *>(nullptr);
+  if (BIO_new_bio_dgram_pair(&ssl_bio, 0, &network_bio, 0) != 1 || ssl_bio == nullptr || network_bio == nullptr) {
+    BIO_free(ssl_bio);
+    BIO_free(network_bio);
+    return ligo::detail::fail<std::unique_ptr<dtls_state>>(dtls_failure("failed to create UDP DTLS BIO pair"));
+  }
+  BIO_up_ref(ssl_bio);
+  SSL_set_bio(ssl.get(), ssl_bio, ssl_bio);
+  if (server) {
+    SSL_set_accept_state(ssl.get());
+  } else {
+    SSL_set_connect_state(ssl.get());
+  }
+  return modulus::success(std::make_unique<dtls_state>(dtls_state{
+    .context = std::move(context),
+    .ssl = std::move(ssl),
+    .network_bio = std::unique_ptr<BIO, bio_deleter>{network_bio},
+    .server = server,
+  }));
+}
+
+auto drain_dtls_output(dtls_state &state) -> std::expected<std::vector<std::string>, modulus::error> {
+  auto output = std::vector<std::string>{};
+  auto *bio = state.network_bio.get();
+  while (BIO_ctrl_pending(bio) > 0) {
+    auto const pending = static_cast<std::size_t>(BIO_ctrl_pending(bio));
+    auto datagram = std::string(std::min(pending, max_datagram_size), '\0');
+    auto const read = BIO_read(bio, datagram.data(), static_cast<std::int32_t>(datagram.size()));
+    if (read <= 0) {
+      return ligo::detail::fail<std::vector<std::string>>(dtls_failure("failed to read UDP DTLS output"));
+    }
+    datagram.resize(static_cast<std::size_t>(read));
+    output.push_back(std::move(datagram));
+  }
+  return modulus::success(std::move(output));
+}
+
+auto append_dtls_output(std::vector<std::string> &target, dtls_state &state) -> std::expected<void, modulus::error> {
+  auto drained = drain_dtls_output(state);
+  if (!drained) {
+    return ligo::detail::fail_void(drained.error());
+  }
+  for (auto &datagram : *drained) {
+    target.push_back(std::move(datagram));
+  }
+  return {};
+}
+
+auto run_dtls_handshake(dtls_state &state, std::vector<std::string> &output) -> std::expected<void, modulus::error> {
+  if (state.handshake_complete) {
+    return {};
+  }
+  state.handshake_started = true;
+  auto const result = SSL_do_handshake(state.ssl.get());
+  if (result == 1) {
+    state.handshake_complete = true;
+    return append_dtls_output(output, state);
+  }
+  auto const error = SSL_get_error(state.ssl.get(), result);
+  if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE) {
+    return append_dtls_output(output, state);
+  }
+  return ligo::detail::fail_void(dtls_failure("UDP DTLS handshake failed"));
+}
+
+auto protect_dtls_payload(dtls_state *state, std::string_view plaintext) -> std::expected<std::vector<std::string>, modulus::error> {
+  if (!state) {
+    return modulus::success(std::vector<std::string>{
+      std::string{plaintext},
+    });
+  }
+  if (plaintext.size() > max_dtls_plaintext_size) {
+    return ligo::detail::fail<std::vector<std::string>>(ligo::detail::error_invalid_argument, "UDP DTLS payload is too large");
+  }
+  auto output = std::vector<std::string>{};
+  auto handshake = run_dtls_handshake(*state, output);
+  if (!handshake) {
+    return ligo::detail::fail<std::vector<std::string>>(handshake.error());
+  }
+  if (!state->handshake_complete) {
+    return modulus::success(std::move(output));
+  }
+  auto const written = SSL_write(state->ssl.get(), plaintext.data(), static_cast<std::int32_t>(plaintext.size()));
+  if (written <= 0) {
+    auto const error = SSL_get_error(state->ssl.get(), written);
+    if (error != SSL_ERROR_WANT_READ && error != SSL_ERROR_WANT_WRITE) {
+      return ligo::detail::fail<std::vector<std::string>>(dtls_failure("UDP DTLS write failed"));
+    }
+  }
+  auto drained = append_dtls_output(output, *state);
+  if (!drained) {
+    return ligo::detail::fail<std::vector<std::string>>(drained.error());
+  }
+  return modulus::success(std::move(output));
+}
+
+auto process_dtls_input(dtls_state *state, std::string_view datagram) -> std::expected<dtls_processed, modulus::error> {
+  auto processed = dtls_processed{};
+  if (!state) {
+    processed.plaintext.push_back(std::string{datagram});
+    return modulus::success(std::move(processed));
+  }
+  auto const written = BIO_write(state->network_bio.get(), datagram.data(), static_cast<std::int32_t>(datagram.size()));
+  if (written <= 0) {
+    return ligo::detail::fail<dtls_processed>(dtls_failure("failed to write UDP DTLS input"));
+  }
+  auto handshake = run_dtls_handshake(*state, processed.wire);
+  if (!handshake) {
+    return ligo::detail::fail<dtls_processed>(handshake.error());
+  }
+  if (!state->handshake_complete) {
+    return modulus::success(std::move(processed));
+  }
+  for (;;) {
+    auto plaintext = std::string(max_dtls_plaintext_size, '\0');
+    auto const read = SSL_read(state->ssl.get(), plaintext.data(), static_cast<std::int32_t>(plaintext.size()));
+    if (read > 0) {
+      plaintext.resize(static_cast<std::size_t>(read));
+      processed.plaintext.push_back(std::move(plaintext));
+      continue;
+    }
+    auto const error = SSL_get_error(state->ssl.get(), read);
+    if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE || error == SSL_ERROR_ZERO_RETURN) {
+      break;
+    }
+    return ligo::detail::fail<dtls_processed>(dtls_failure("UDP DTLS read failed"));
+  }
+  auto drained = append_dtls_output(processed.wire, *state);
+  if (!drained) {
+    return ligo::detail::fail<dtls_processed>(drained.error());
+  }
+  return modulus::success(std::move(processed));
+}
+
+auto ssl_ctx_deleter::operator()(SSL_CTX *context) const noexcept -> void {
+  SSL_CTX_free(context);
+}
+
+auto ssl_deleter::operator()(SSL *ssl) const noexcept -> void {
+  SSL_free(ssl);
+}
+
+auto bio_deleter::operator()(BIO *bio) const noexcept -> void {
+  BIO_free(bio);
+}
+
+}

src/detail/dtls.hpp

@@ -0,0 +1,46 @@
+#pragma once
+
+#include <cstdint>
+#include <expected>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <vector>
+#include <openssl/types.h>
+#include <modulus.hpp>
+#include <ligo/config.hpp>
+
+
+namespace ligo::detail {
+
+struct ssl_ctx_deleter {
+  auto operator()(SSL_CTX *context) const noexcept -> void;
+};
+
+struct ssl_deleter {
+  auto operator()(SSL *ssl) const noexcept -> void;
+};
+
+struct bio_deleter {
+  auto operator()(BIO *bio) const noexcept -> void;
+};
+
+struct dtls_state {
+  std::unique_ptr<SSL_CTX, ssl_ctx_deleter> context{};
+  std::unique_ptr<SSL, ssl_deleter> ssl{};
+  std::unique_ptr<BIO, bio_deleter> network_bio{};
+  std::bool_t server{};
+  std::bool_t handshake_started{};
+  std::bool_t handshake_complete{};
+};
+
+struct dtls_processed {
+  std::vector<std::string> wire{};
+  std::vector<std::string> plaintext{};
+};
+
+auto make_dtls_state(ligo::security_config const &config, std::bool_t server) -> std::expected<std::unique_ptr<dtls_state>, modulus::error>;
+auto protect_dtls_payload(dtls_state *state, std::string_view plaintext) -> std::expected<std::vector<std::string>, modulus::error>;
+auto process_dtls_input(dtls_state *state, std::string_view datagram) -> std::expected<dtls_processed, modulus::error>;
+
+}

src/detail/udp_protocol.cpp

@@ -0,0 +1,64 @@
+#include "udp_protocol.hpp"
+
+#include <chrono>
+#include <utility>
+#include <boost/json/serialize.hpp>
+#include <ligo/detail/errors.hpp>
+#include <ligo/detail/transport.hpp>
+
+
+namespace ligo::detail {
+
+auto uint_member(boost::json::object const &object, std::string_view name) -> std::optional<std::uint64_t> {
+  auto found = object.find(name);
+  if (found == object.end() || !found->value().is_number()) {
+    return std::nullopt;
+  }
+  if (found->value().is_int64() && found->value().as_int64() < 0) {
+    return std::nullopt;
+  }
+  if (found->value().is_double()) {
+    return std::nullopt;
+  }
+  return found->value().is_uint64() ? found->value().as_uint64() : static_cast<std::uint64_t>(found->value().as_int64());
+}
+
+auto string_member(boost::json::object const &object, std::string_view name) -> std::string {
+  auto found = object.find(name);
+  if (found == object.end() || !found->value().is_string()) {
+    return {};
+  }
+  return std::string{found->value().as_string()};
+}
+
+auto make_data(std::uint64_t session_id, std::uint64_t sequence, boost::json::value payload) -> std::string {
+  auto envelope = boost::json::object{};
+  envelope["kind"] = "data";
+  envelope["session"] = session_id;
+  envelope["sequence"] = sequence;
+  envelope["payload"] = std::move(payload);
+  return boost::json::serialize(envelope);
+}
+
+auto make_ack(std::uint64_t session_id, std::uint64_t sequence) -> std::string {
+  auto envelope = boost::json::object{};
+  envelope["kind"] = "ack";
+  envelope["session"] = session_id;
+  envelope["sequence"] = sequence;
+  return boost::json::serialize(envelope);
+}
+
+auto validate_config(ligo::udp_reliability_config const &config) -> std::expected<void, modulus::error> {
+  if (config.resend_interval <= std::chrono::milliseconds{0}) {
+    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP resend_interval must be positive"));
+  }
+  if (config.max_in_flight == 0) {
+    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP max_in_flight must be positive"));
+  }
+  if (config.session_timeout <= std::chrono::milliseconds{0}) {
+    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP session_timeout must be positive"));
+  }
+  return {};
+}
+
+}

src/detail/udp_protocol.hpp

@@ -0,0 +1,22 @@
+#pragma once
+
+#include <cstdint>
+#include <expected>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <boost/json/object.hpp>
+#include <boost/json/value.hpp>
+#include <modulus.hpp>
+#include <ligo/config.hpp>
+
+
+namespace ligo::detail {
+
+auto uint_member(boost::json::object const &object, std::string_view name) -> std::optional<std::uint64_t>;
+auto string_member(boost::json::object const &object, std::string_view name) -> std::string;
+auto make_data(std::uint64_t session_id, std::uint64_t sequence, boost::json::value payload) -> std::string;
+auto make_ack(std::uint64_t session_id, std::uint64_t sequence) -> std::string;
+auto validate_config(ligo::udp_reliability_config const &config) -> std::expected<void, modulus::error>;
+
+}

src/http.cpp

@@ -1,6 +1,9 @@
 #include <ligo/http.hpp>
 
 #include <atomic>
+#include <algorithm>
+#include <cctype>
+#include <stdexcept>
 #include <cassert>
 #include <chrono>
 #include <cstdint>
@@ -118,6 +121,67 @@ auto host_header(std::string const &host, std::uint16_t port) -> std::string {
   return value;
 }
 
+auto lower(std::string value) -> std::string {
+  std::transform(value.begin(), value.end(), value.begin(), [](unsigned char character) { return static_cast<char>(std::tolower(character)); });
+  return value;
+}
+
+auto decode_query(std::string_view value) -> std::string {
+  auto output = std::string{};
+  auto hex = [](char character) -> int {
+    if (character >= '0' && character <= '9') {
+      return character - '0';
+    }
+    if (character >= 'a' && character <= 'f') {
+      return character - 'a' + 10;
+    }
+    if (character >= 'A' && character <= 'F') {
+      return character - 'A' + 10;
+    }
+    return -1;
+  };
+  for (std::size_t index = 0; index < value.size(); ++index) {
+    if (value[index] == '%') {
+      if (index + 2 >= value.size() || hex(value[index + 1]) < 0 || hex(value[index + 2]) < 0) {
+        throw std::invalid_argument("Invalid query escape");
+      }
+      auto character = static_cast<char>(hex(value[index + 1]) * 16 + hex(value[index + 2]));
+      if (character == '\0') {
+        throw std::invalid_argument("Invalid query character");
+      }
+      output += character;
+      index += 2;
+    } else {
+      output += value[index] == '+' ? ' ' : value[index];
+    }
+  }
+  return output;
+}
+
+auto parse_query(ligo::request_context &context) -> void {
+  auto position = context.route.find('?');
+  if (position == std::string::npos) {
+    return;
+  }
+  auto query = context.route.substr(position + 1);
+  context.route.resize(position);
+  auto offset = std::size_t{};
+  while (offset < query.size()) {
+    auto end = query.find('&', offset);
+    auto part = std::string_view{query}.substr(offset, end == std::string::npos ? query.size() - offset : end - offset);
+    auto equal = part.find('=');
+    auto key = decode_query(part.substr(0, equal));
+    auto value = equal == std::string_view::npos ? std::string{} : decode_query(part.substr(equal + 1));
+    if (key.empty() || !context.query.emplace(std::move(key), std::move(value)).second) {
+      throw std::invalid_argument("Empty or duplicate query key");
+    }
+    if (end == std::string::npos) {
+      break;
+    }
+    offset = end + 1;
+  }
+}
+
 auto remote_peer(boost::beast::tcp_stream &stream) -> std::string {
   auto endpoint_error = boost::system::error_code{};
   auto remote_endpoint = stream.socket().remote_endpoint(endpoint_error);
@@ -196,9 +260,33 @@ auto handle_session(STREAM &stream, std::shared_ptr<ligo::service_router const>
     .method = std::string{http_request.method_string()},
     .route = std::string{http_request.target()},
   };
+  for (auto const &field : http_request.base()) {
+    auto name = lower(std::string{field.name_string()});
+    auto value = std::string{field.value()};
+    if (context.headers.contains(name)) {
+      context.headers[name] += (name == "cookie" ? "; " : ", ") + value;
+    } else {
+      context.headers.emplace(std::move(name), std::move(value));
+    }
+  }
+  auto query_error = std::string{};
+  try {
+    if (options.parse_query) {
+      parse_query(context);
+    }
+  } catch (std::invalid_argument const &error) {
+    query_error = error.what();
+  }
+  if (!query_error.empty()) {
+    co_await write_http_failure_response(stream, ligo::detail::make_error(ligo::detail::error_invalid_argument, query_error), options.timeout, options.on_error);
+    co_return;
+  }
   auto result = co_await ligo::detail::call_response_handler(*router, std::move(context), std::move(payload));
   if (result) {
-    response.result(boost::beast::http::status::ok);
+    response.result(result->status >= 200 && result->status <= 599 ? result->status : 500);
+    for (auto const &[name, value] : result->headers) {
+      response.insert(name, value);
+    }
     response.set(boost::beast::http::field::content_type, result->content_type);
     response.body() = std::move(result->body);
   } else {
@@ -326,12 +414,18 @@ auto make_http_request_message(ligo::http::endpoint const &endpoint, ligo::http:
     http_request.set(boost::beast::http::field::content_type, "application/json");
     http_request.body() = boost::json::serialize(request.payload);
   }
+  if (request.raw_body) {
+    http_request.body() = *request.raw_body;
+  }
+  for (auto const &[name, value] : request.headers) {
+    http_request.set(name, value);
+  }
   http_request.prepare_payload();
   return http_request;
 }
 
 template <typename STREAM>
-auto read_http_response(STREAM &stream, std::chrono::milliseconds timeout, std::uint64_t max_body_size) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
+auto read_http_response(STREAM &stream, std::chrono::milliseconds timeout, std::uint64_t max_body_size) -> boost::asio::awaitable<std::expected<ligo::http::response, modulus::error>> {
   auto buffer = boost::beast::flat_buffer{};
   auto parser = boost::beast::http::response_parser<boost::beast::http::string_body>{};
   parser.body_limit(max_body_size);
@@ -342,36 +436,42 @@ auto read_http_response(STREAM &stream, std::chrono::milliseconds timeout, std::
   if (read_error) {
     if (http_body_limit_exceeded(read_error)) {
       auto failure_status = http_body_limit_status();
-      co_return ligo::detail::fail<boost::json::value>(failure_status);
+      co_return ligo::detail::fail<ligo::http::response>(failure_status);
     }
     auto failure_status = ligo::detail::operation_failure(read_error, read_deadline, "HTTP request timed out");
-    co_return ligo::detail::fail<boost::json::value>(failure_status);
+    co_return ligo::detail::fail<ligo::http::response>(failure_status);
   }
   auto http_response = parser.release();
-  auto parse_error = boost::system::error_code{};
-  auto body = boost::json::parse(http_response.body(), parse_error);
-  if (parse_error) {
-    co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_invalid_argument, parse_error.message());
+  auto response = ligo::http::response{};
+  response.status = static_cast<std::uint16_t>(http_response.result_int());
+  response.body = std::move(http_response.body());
+  for (auto const &field : http_response.base()) {
+    auto name = lower(std::string{field.name_string()});
+    if (response.headers.contains(name)) {
+      response.headers[name] += ", " + std::string{field.value()};
+    } else {
+      response.headers.emplace(std::move(name), std::string{field.value()});
+    }
   }
-  co_return ligo::detail::unwrap_response_envelope(std::move(body), "HTTP");
+  co_return response;
 }
 
-auto call_https_endpoint(ligo::http::endpoint endpoint, boost::beast::http::request<boost::beast::http::string_body> http_request, ligo::http::client_options options) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
+auto call_https_endpoint(ligo::http::endpoint endpoint, boost::beast::http::request<boost::beast::http::string_body> http_request, ligo::http::client_options options) -> boost::asio::awaitable<std::expected<ligo::http::response, modulus::error>> {
   auto context = ligo::make_tls_client_context(options.security);
   if (!context) {
-    co_return ligo::detail::fail<boost::json::value>(context.error());
+    co_return ligo::detail::fail<ligo::http::response>(context.error());
   }
   auto executor = co_await boost::asio::this_coro::executor;
   auto endpoints = co_await ligo::detail::resolve_tcp_endpoints(executor, endpoint.host, endpoint.port, options.timeout, "HTTP request timed out");
   if (!endpoints) {
-    co_return ligo::detail::fail<boost::json::value>(endpoints.error());
+    co_return ligo::detail::fail<ligo::http::response>(endpoints.error());
   }
   auto tcp_stream = boost::beast::tcp_stream{executor};
   tcp_stream.expires_after(options.timeout);
   auto [connect_error, _] = co_await tcp_stream.async_connect(*endpoints, boost::asio::as_tuple(boost::asio::use_awaitable));
   if (connect_error) {
     auto failure_status = ligo::detail::operation_failure(connect_error, "HTTP request timed out");
-    co_return ligo::detail::fail<boost::json::value>(failure_status);
+    co_return ligo::detail::fail<ligo::http::response>(failure_status);
   }
   auto stream = boost::asio::ssl::stream<boost::asio::ip::tcp::socket>{
     tcp_stream.release_socket(),
@@ -379,11 +479,11 @@ auto call_https_endpoint(ligo::http::endpoint endpoint, boost::beast::http::requ
   };
   auto verification = ligo::configure_tls_client_stream(stream, endpoint.host, options.security);
   if (!verification) {
-    co_return ligo::detail::fail<boost::json::value>(verification.error());
+    co_return ligo::detail::fail<ligo::http::response>(verification.error());
   }
   auto handshake = co_await ligo::detail::async_tls_handshake(stream, boost::asio::ssl::stream_base::client, options.timeout, "HTTP TLS handshake timed out");
   if (!handshake) {
-    co_return ligo::detail::fail<boost::json::value>(handshake.error());
+    co_return ligo::detail::fail<ligo::http::response>(handshake.error());
   }
   auto write_deadline = ligo::detail::start_stream_deadline(stream, options.timeout);
   auto [write_error, written] = co_await boost::beast::http::async_write(stream, http_request, ligo::detail::deadline_token(write_deadline));
@@ -391,36 +491,36 @@ auto call_https_endpoint(ligo::http::endpoint endpoint, boost::beast::http::requ
   static_cast<void>(written);
   if (write_error) {
     auto failure_status = ligo::detail::operation_failure(write_error, write_deadline, "HTTP request timed out");
-    co_return ligo::detail::fail<boost::json::value>(failure_status);
+    co_return ligo::detail::fail<ligo::http::response>(failure_status);
   }
   auto response = co_await read_http_response(stream, options.timeout, options.max_body_size);
   static_cast<void>(co_await ligo::detail::async_tls_shutdown(stream, options.timeout, "HTTP TLS shutdown timed out"));
   co_return response;
 }
 
-auto call_http_endpoint(ligo::http::endpoint endpoint, std::shared_ptr<ligo::http::request const> request, ligo::http::client_options options) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
+auto call_http_endpoint(ligo::http::endpoint endpoint, std::shared_ptr<ligo::http::request const> request, ligo::http::client_options options) -> boost::asio::awaitable<std::expected<ligo::http::response, modulus::error>> {
   assert(request != nullptr);
   modulus::log_debug("ligo HTTP client call to " + endpoint.host + ":" + std::to_string(endpoint.port));
   auto valid_timeout = ligo::detail::validate_positive_timeout(options.timeout, "HTTP timeout");
   if (!valid_timeout) {
-    co_return ligo::detail::fail<boost::json::value>(valid_timeout.error());
+    co_return ligo::detail::fail<ligo::http::response>(valid_timeout.error());
   }
   if (endpoint.port == 0) {
-    co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_invalid_argument, "HTTP port must be non-zero");
+    co_return ligo::detail::fail<ligo::http::response>(ligo::detail::error_invalid_argument, "HTTP port must be non-zero");
   }
   if (request->target.empty()) {
-    co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_invalid_argument, "HTTP target must not be empty");
+    co_return ligo::detail::fail<ligo::http::response>(ligo::detail::error_invalid_argument, "HTTP target must not be empty");
   }
   if (request->target.front() != '/') {
-    co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_invalid_argument, "HTTP target must start with /");
+    co_return ligo::detail::fail<ligo::http::response>(ligo::detail::error_invalid_argument, "HTTP target must start with /");
   }
   auto verb = http_verb(request->method);
   if (!verb) {
-    co_return ligo::detail::fail<boost::json::value>(verb.error());
+    co_return ligo::detail::fail<ligo::http::response>(verb.error());
   }
   if (options.security.mode == ligo::security_mode::DTLS) {
     auto status = ligo::unsupported_security("HTTP client", options.security.mode);
-    co_return ligo::detail::fail<boost::json::value>(status);
+    co_return ligo::detail::fail<ligo::http::response>(status);
   }
   auto http_request = make_http_request_message(endpoint, *request, *verb);
   if (options.security.mode == ligo::security_mode::TLS) {
@@ -430,25 +530,38 @@ auto call_http_endpoint(ligo::http::endpoint endpoint, std::shared_ptr<ligo::htt
   auto stream = boost::beast::tcp_stream{executor};
   auto endpoints = co_await ligo::detail::resolve_tcp_endpoints(executor, endpoint.host, endpoint.port, options.timeout, "HTTP request timed out");
   if (!endpoints) {
-    co_return ligo::detail::fail<boost::json::value>(endpoints.error());
+    co_return ligo::detail::fail<ligo::http::response>(endpoints.error());
   }
   stream.expires_after(options.timeout);
   auto [connect_error, _] = co_await stream.async_connect(*endpoints, boost::asio::as_tuple(boost::asio::use_awaitable));
   if (connect_error) {
     auto failure_status = ligo::detail::operation_failure(connect_error, "HTTP request timed out");
-    co_return ligo::detail::fail<boost::json::value>(failure_status);
+    co_return ligo::detail::fail<ligo::http::response>(failure_status);
   }
   stream.expires_after(options.timeout);
   auto [write_error, written] = co_await boost::beast::http::async_write(stream, http_request, boost::asio::as_tuple(boost::asio::use_awaitable));
   static_cast<void>(written);
   if (write_error) {
     auto failure_status = ligo::detail::operation_failure(write_error, "HTTP request timed out");
-    co_return ligo::detail::fail<boost::json::value>(failure_status);
+    co_return ligo::detail::fail<ligo::http::response>(failure_status);
   }
   stream.expires_after(options.timeout);
   co_return co_await read_http_response(stream, options.timeout, options.max_body_size);
 }
 
+auto unwrap_http_response(boost::asio::awaitable<std::expected<ligo::http::response, modulus::error>> operation) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
+  auto response = co_await std::move(operation);
+  if (!response) {
+    co_return std::unexpected(response.error());
+  }
+  auto error = boost::system::error_code{};
+  auto body = boost::json::parse(response->body, error);
+  if (error) {
+    co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_invalid_argument, error.message());
+  }
+  co_return ligo::detail::unwrap_response_envelope(std::move(body), "HTTP");
+}
+
 http_server_state::http_server_state(boost::asio::any_io_executor executor, ligo::http::endpoint endpoint, ligo::service_router router, ligo::http::server_options options)
     : ligo::detail::stream_listener{std::move(executor), std::move(router)}, endpoint{std::move(endpoint)}, options{options} {}
 
@@ -458,6 +571,11 @@ ligo::http::client::client(ligo::http::endpoint endpoint, ligo::http::client_opt
     : endpoint_{std::move(endpoint)}, options_{options} {}
 
 auto ligo::http::client::call(ligo::http::request const &request) const -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
+  // Capture inputs now: the client and request may expire before the coroutine runs.
+  return unwrap_http_response(exchange(request));
+}
+
+auto ligo::http::client::exchange(ligo::http::request const &request) const -> boost::asio::awaitable<std::expected<ligo::http::response, modulus::error>> {
   return call_http_endpoint(endpoint_, std::make_shared<ligo::http::request const>(request), options_);
 }
 

src/udp.cpp

@@ -1,3 +1,5 @@
+#include "detail/dtls.hpp"
+#include "detail/udp_protocol.hpp"
 #include <ligo/udp.hpp>
 
 #include <algorithm>
@@ -51,41 +53,15 @@
 
 namespace {
 
-struct ssl_ctx_deleter {
-  auto operator()(SSL_CTX *context) const noexcept -> void;
-};
-
-struct ssl_deleter {
-  auto operator()(SSL *ssl) const noexcept -> void;
-};
-
-struct bio_deleter {
-  auto operator()(BIO *bio) const noexcept -> void;
-};
-
-struct dtls_state {
-  std::unique_ptr<SSL_CTX, ssl_ctx_deleter> context{};
-  std::unique_ptr<SSL, ssl_deleter> ssl{};
-  std::unique_ptr<BIO, bio_deleter> network_bio{};
-  std::bool_t server{};
-  std::bool_t handshake_started{};
-  std::bool_t handshake_complete{};
-};
-
-struct dtls_processed {
-  std::vector<std::string> wire{};
-  std::vector<std::string> plaintext{};
-};
-
 struct reliable_channel_state {
-  reliable_channel_state(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint local_endpoint, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::unique_ptr<dtls_state> dtls);
+  reliable_channel_state(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint local_endpoint, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::unique_ptr<ligo::detail::dtls_state> dtls);
   std::unique_ptr<boost::asio::ip::udp::socket> socket{};
   boost::asio::strand<boost::asio::ip::udp::socket::executor_type> strand{boost::asio::make_strand(socket->get_executor())};
   boost::asio::ip::udp::endpoint local_endpoint{};
   boost::asio::ip::udp::endpoint peer{};
   ligo::udp_reliability_config config{};
   ligo::security_config security{};
-  std::unique_ptr<dtls_state> dtls{};
+  std::unique_ptr<ligo::detail::dtls_state> dtls{};
   std::uint64_t session_id{1};
   std::uint64_t next_send_sequence{1};
   std::uint64_t next_receive_sequence{1};
@@ -107,300 +83,11 @@ struct reliable_channel_state {
 };
 
 constexpr auto max_datagram_size = std::size_t{65507};
-constexpr auto max_dtls_plaintext_size = std::size_t{12000};
 
 auto same_peer(boost::asio::ip::udp::endpoint const &left, boost::asio::ip::udp::endpoint const &right) -> std::bool_t {
   return left.address() == right.address() && left.port() == right.port();
 }
 
-auto uint_member(boost::json::object const &object, std::string_view name) -> std::optional<std::uint64_t> {
-  auto found = object.find(name);
-  if (found == object.end() || !found->value().is_number()) {
-    return std::nullopt;
-  }
-  if (found->value().is_int64() && found->value().as_int64() < 0) {
-    return std::nullopt;
-  }
-  if (found->value().is_double()) {
-    return std::nullopt;
-  }
-  return found->value().is_uint64() ? found->value().as_uint64() : static_cast<std::uint64_t>(found->value().as_int64());
-}
-
-auto string_member(boost::json::object const &object, std::string_view name) -> std::string {
-  auto found = object.find(name);
-  if (found == object.end() || !found->value().is_string()) {
-    return {};
-  }
-  return std::string{found->value().as_string()};
-}
-
-auto make_data(std::uint64_t session_id, std::uint64_t sequence, boost::json::value payload) -> std::string {
-  auto envelope = boost::json::object{};
-  envelope["kind"] = "data";
-  envelope["session"] = session_id;
-  envelope["sequence"] = sequence;
-  envelope["payload"] = std::move(payload);
-  return boost::json::serialize(envelope);
-}
-
-auto make_ack(std::uint64_t session_id, std::uint64_t sequence) -> std::string {
-  auto envelope = boost::json::object{};
-  envelope["kind"] = "ack";
-  envelope["session"] = session_id;
-  envelope["sequence"] = sequence;
-  return boost::json::serialize(envelope);
-}
-
-auto validate_config(ligo::udp_reliability_config const &config) -> std::expected<void, modulus::error> {
-  if (config.resend_interval <= std::chrono::milliseconds{0}) {
-    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP resend_interval must be positive"));
-  }
-  if (config.max_in_flight == 0) {
-    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP max_in_flight must be positive"));
-  }
-  if (config.session_timeout <= std::chrono::milliseconds{0}) {
-    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP session_timeout must be positive"));
-  }
-  return {};
-}
-
-auto dtls_failure(std::int32_t code, std::string message) -> modulus::error {
-  auto const error = ERR_get_error();
-  if (error != 0) {
-    message += ": ";
-    message += ERR_error_string(error, nullptr);
-  }
-  return ligo::detail::make_error(code, std::move(message));
-}
-
-auto dtls_failure(std::string message) -> modulus::error {
-  return dtls_failure(ligo::detail::error_unavailable, std::move(message));
-}
-
-auto validate_security(ligo::security_config const &config, std::bool_t server) -> std::expected<void, modulus::error> {
-  if (config.mode == ligo::security_mode::TLS) {
-    return ligo::detail::fail_void(ligo::unsupported_security("UDP", config.mode));
-  }
-  if (config.mode != ligo::security_mode::DTLS) {
-    return {};
-  }
-  if (config.tls_context) {
-    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS does not accept a TLS context"));
-  }
-  if (server && (config.certificate_chain_file.empty() || config.private_key_file.empty())) {
-    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS server requires certificate chain/private key files"));
-  }
-  if (!config.certificate_chain_file.empty() != !config.private_key_file.empty()) {
-    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS certificate chain and private key must both be configured"));
-  }
-  return {};
-}
-
-auto load_dtls_verify_paths(SSL_CTX *context, ligo::security_config const &config) -> std::expected<void, modulus::error> {
-  if (config.verify == ligo::verify_mode::NONE) {
-    return {};
-  }
-  if (!config.ca_file.empty()) {
-    if (SSL_CTX_load_verify_locations(context, config.ca_file.c_str(), nullptr) != 1) {
-      return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS CA file"));
-    }
-    return {};
-  }
-  if (SSL_CTX_set_default_verify_paths(context) != 1) {
-    return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS default verify paths"));
-  }
-  return {};
-}
-
-auto load_dtls_identity(SSL_CTX *context, ligo::security_config const &config) -> std::expected<void, modulus::error> {
-  if (config.certificate_chain_file.empty() && config.private_key_file.empty()) {
-    return {};
-  }
-  if (SSL_CTX_use_certificate_chain_file(context, config.certificate_chain_file.c_str()) != 1) {
-    return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS certificate chain"));
-  }
-  if (SSL_CTX_use_PrivateKey_file(context, config.private_key_file.c_str(), SSL_FILETYPE_PEM) != 1) {
-    return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS private key"));
-  }
-  if (SSL_CTX_check_private_key(context) != 1) {
-    return ligo::detail::fail_void(dtls_failure(ligo::detail::error_invalid_argument, "UDP DTLS private key does not match certificate"));
-  }
-  return {};
-}
-
-auto make_dtls_state(ligo::security_config const &config, std::bool_t server) -> std::expected<std::unique_ptr<dtls_state>, modulus::error> {
-  if (config.mode != ligo::security_mode::DTLS) {
-    return modulus::success(std::unique_ptr<dtls_state>{});
-  }
-  auto valid = validate_security(config, server);
-  if (!valid) {
-    return ligo::detail::fail<std::unique_ptr<dtls_state>>(valid.error());
-  }
-  auto context = std::unique_ptr<SSL_CTX, ssl_ctx_deleter>{SSL_CTX_new(server ? DTLS_server_method() : DTLS_client_method())};
-  if (!context) {
-    return ligo::detail::fail<std::unique_ptr<dtls_state>>(ligo::detail::error_unavailable, "failed to create UDP DTLS context");
-  }
-  SSL_CTX_set_options(context.get(), SSL_OP_NO_COMPRESSION | SSL_OP_NO_QUERY_MTU);
-  if (SSL_CTX_set_min_proto_version(context.get(), DTLS1_2_VERSION) != 1) {
-    return ligo::detail::fail<std::unique_ptr<dtls_state>>(dtls_failure("failed to configure UDP DTLS minimum protocol"));
-  }
-  auto const verify_peer = config.verify == ligo::verify_mode::PEER || (!server && config.verify == ligo::verify_mode::DEFAULTED);
-  auto const verify = verify_peer ? SSL_VERIFY_PEER | (server ? SSL_VERIFY_FAIL_IF_NO_PEER_CERT : 0) : SSL_VERIFY_NONE;
-  SSL_CTX_set_verify(context.get(), verify, nullptr);
-  auto verify_paths = load_dtls_verify_paths(context.get(), config);
-  if (!verify_paths) {
-    return ligo::detail::fail<std::unique_ptr<dtls_state>>(verify_paths.error());
-  }
-  auto identity = load_dtls_identity(context.get(), config);
-  if (!identity) {
-    return ligo::detail::fail<std::unique_ptr<dtls_state>>(identity.error());
-  }
-  auto ssl = std::unique_ptr<SSL, ssl_deleter>{SSL_new(context.get())};
-  if (!ssl) {
-    return ligo::detail::fail<std::unique_ptr<dtls_state>>(ligo::detail::error_unavailable, "failed to create UDP DTLS session");
-  }
-  if (!server && verify_peer && !config.peer_name.empty()) {
-    if (X509_VERIFY_PARAM_set1_host(SSL_get0_param(ssl.get()), config.peer_name.c_str(), 0) != 1) {
-      return ligo::detail::fail<std::unique_ptr<dtls_state>>(dtls_failure("failed to configure UDP DTLS peer identity"));
-    }
-  }
-  SSL_set_mtu(ssl.get(), 1200);
-  auto *ssl_bio = static_cast<BIO *>(nullptr);
-  auto *network_bio = static_cast<BIO *>(nullptr);
-  if (BIO_new_bio_dgram_pair(&ssl_bio, 0, &network_bio, 0) != 1 || ssl_bio == nullptr || network_bio == nullptr) {
-    BIO_free(ssl_bio);
-    BIO_free(network_bio);
-    return ligo::detail::fail<std::unique_ptr<dtls_state>>(dtls_failure("failed to create UDP DTLS BIO pair"));
-  }
-  BIO_up_ref(ssl_bio);
-  SSL_set_bio(ssl.get(), ssl_bio, ssl_bio);
-  if (server) {
-    SSL_set_accept_state(ssl.get());
-  } else {
-    SSL_set_connect_state(ssl.get());
-  }
-  return modulus::success(std::make_unique<dtls_state>(dtls_state{
-    .context = std::move(context),
-    .ssl = std::move(ssl),
-    .network_bio = std::unique_ptr<BIO, bio_deleter>{network_bio},
-    .server = server,
-  }));
-}
-
-auto drain_dtls_output(dtls_state &state) -> std::expected<std::vector<std::string>, modulus::error> {
-  auto output = std::vector<std::string>{};
-  auto *bio = state.network_bio.get();
-  while (BIO_ctrl_pending(bio) > 0) {
-    auto const pending = static_cast<std::size_t>(BIO_ctrl_pending(bio));
-    auto datagram = std::string(std::min(pending, max_datagram_size), '\0');
-    auto const read = BIO_read(bio, datagram.data(), static_cast<std::int32_t>(datagram.size()));
-    if (read <= 0) {
-      return ligo::detail::fail<std::vector<std::string>>(dtls_failure("failed to read UDP DTLS output"));
-    }
-    datagram.resize(static_cast<std::size_t>(read));
-    output.push_back(std::move(datagram));
-  }
-  return modulus::success(std::move(output));
-}
-
-auto append_dtls_output(std::vector<std::string> &target, dtls_state &state) -> std::expected<void, modulus::error> {
-  auto drained = drain_dtls_output(state);
-  if (!drained) {
-    return ligo::detail::fail_void(drained.error());
-  }
-  for (auto &datagram : *drained) {
-    target.push_back(std::move(datagram));
-  }
-  return {};
-}
-
-auto run_dtls_handshake(dtls_state &state, std::vector<std::string> &output) -> std::expected<void, modulus::error> {
-  if (state.handshake_complete) {
-    return {};
-  }
-  state.handshake_started = true;
-  auto const result = SSL_do_handshake(state.ssl.get());
-  if (result == 1) {
-    state.handshake_complete = true;
-    return append_dtls_output(output, state);
-  }
-  auto const error = SSL_get_error(state.ssl.get(), result);
-  if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE) {
-    return append_dtls_output(output, state);
-  }
-  return ligo::detail::fail_void(dtls_failure("UDP DTLS handshake failed"));
-}
-
-auto protect_dtls_payload(reliable_channel_state &state, std::string_view plaintext) -> std::expected<std::vector<std::string>, modulus::error> {
-  if (!state.dtls) {
-    return modulus::success(std::vector<std::string>{
-      std::string{plaintext},
-    });
-  }
-  if (plaintext.size() > max_dtls_plaintext_size) {
-    return ligo::detail::fail<std::vector<std::string>>(ligo::detail::error_invalid_argument, "UDP DTLS payload is too large");
-  }
-  auto output = std::vector<std::string>{};
-  auto handshake = run_dtls_handshake(*state.dtls, output);
-  if (!handshake) {
-    return ligo::detail::fail<std::vector<std::string>>(handshake.error());
-  }
-  if (!state.dtls->handshake_complete) {
-    return modulus::success(std::move(output));
-  }
-  auto const written = SSL_write(state.dtls->ssl.get(), plaintext.data(), static_cast<std::int32_t>(plaintext.size()));
-  if (written <= 0) {
-    auto const error = SSL_get_error(state.dtls->ssl.get(), written);
-    if (error != SSL_ERROR_WANT_READ && error != SSL_ERROR_WANT_WRITE) {
-      return ligo::detail::fail<std::vector<std::string>>(dtls_failure("UDP DTLS write failed"));
-    }
-  }
-  auto drained = append_dtls_output(output, *state.dtls);
-  if (!drained) {
-    return ligo::detail::fail<std::vector<std::string>>(drained.error());
-  }
-  return modulus::success(std::move(output));
-}
-
-auto process_dtls_input(reliable_channel_state &state, std::string_view datagram) -> std::expected<dtls_processed, modulus::error> {
-  auto processed = dtls_processed{};
-  if (!state.dtls) {
-    processed.plaintext.push_back(std::string{datagram});
-    return modulus::success(std::move(processed));
-  }
-  auto const written = BIO_write(state.dtls->network_bio.get(), datagram.data(), static_cast<std::int32_t>(datagram.size()));
-  if (written <= 0) {
-    return ligo::detail::fail<dtls_processed>(dtls_failure("failed to write UDP DTLS input"));
-  }
-  auto handshake = run_dtls_handshake(*state.dtls, processed.wire);
-  if (!handshake) {
-    return ligo::detail::fail<dtls_processed>(handshake.error());
-  }
-  if (!state.dtls->handshake_complete) {
-    return modulus::success(std::move(processed));
-  }
-  for (;;) {
-    auto plaintext = std::string(max_dtls_plaintext_size, '\0');
-    auto const read = SSL_read(state.dtls->ssl.get(), plaintext.data(), static_cast<std::int32_t>(plaintext.size()));
-    if (read > 0) {
-      plaintext.resize(static_cast<std::size_t>(read));
-      processed.plaintext.push_back(std::move(plaintext));
-      continue;
-    }
-    auto const error = SSL_get_error(state.dtls->ssl.get(), read);
-    if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE || error == SSL_ERROR_ZERO_RETURN) {
-      break;
-    }
-    return ligo::detail::fail<dtls_processed>(dtls_failure("UDP DTLS read failed"));
-  }
-  auto drained = append_dtls_output(processed.wire, *state.dtls);
-  if (!drained) {
-    return ligo::detail::fail<dtls_processed>(drained.error());
-  }
-  return modulus::success(std::move(processed));
-}
-
 auto channel_state(ligo::detail::state_handle const &handle) -> std::shared_ptr<reliable_channel_state> {
   return handle.as<reliable_channel_state>();
 }
@@ -432,8 +119,8 @@ auto send_wire_datagrams(std::shared_ptr<reliable_channel_state> const &state, s
 }
 
 auto send_ack(std::shared_ptr<reliable_channel_state> const &state, std::uint64_t sequence) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
-  auto ack_body = make_ack(state->session_id, sequence);
-  auto datagrams = protect_dtls_payload(*state, ack_body);
+  auto ack_body = ligo::detail::make_ack(state->session_id, sequence);
+  auto datagrams = ligo::detail::protect_dtls_payload(state->dtls.get(), ack_body);
   if (!datagrams) {
     co_return ligo::detail::fail_void(datagrams.error());
   }
@@ -487,10 +174,10 @@ auto receive_loop(std::shared_ptr<reliable_channel_state> state) -> boost::asio:
     if (size > max_datagram_size || !same_peer(state->receive_sender, state->peer)) {
       continue;
     }
-    auto processed = process_dtls_input(*state, std::string_view{
-                                                  state->receive_buffer.data(),
-                                                  size,
-                                                });
+    auto processed = ligo::detail::process_dtls_input(state->dtls.get(), std::string_view{
+                                                                           state->receive_buffer.data(),
+                                                                           size,
+                                                                         });
     if (!processed) {
       state->receive_error = processed.error();
       notify_send_waiters(state);
@@ -511,12 +198,12 @@ auto receive_loop(std::shared_ptr<reliable_channel_state> state) -> boost::asio:
         continue;
       }
       auto const &object = envelope.as_object();
-      auto session = uint_member(object, "session");
-      auto sequence = uint_member(object, "sequence");
+      auto session = ligo::detail::uint_member(object, "session");
+      auto sequence = ligo::detail::uint_member(object, "sequence");
       if (!session.has_value() || *session != state->session_id || !sequence.has_value()) {
         continue;
       }
-      auto const kind = string_member(object, "kind");
+      auto const kind = ligo::detail::string_member(object, "kind");
       if (kind == "ack") {
         if (state->active_send_sequences.contains(*sequence)) {
           state->acked.insert(*sequence);
@@ -617,11 +304,11 @@ auto udp_endpoint_from_parts(ligo::detail::endpoint_parts const &parts) -> std::
 }
 
 auto open_state(boost::asio::any_io_executor executor, boost::asio::ip::udp::endpoint local, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::bool_t server) -> std::expected<std::shared_ptr<reliable_channel_state>, modulus::error> {
-  auto valid = validate_config(config);
+  auto valid = ligo::detail::validate_config(config);
   if (!valid) {
     return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(valid.error());
   }
-  auto dtls = make_dtls_state(security, server);
+  auto dtls = ligo::detail::make_dtls_state(security, server);
   if (!dtls) {
     return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(dtls.error());
   }
@@ -646,11 +333,11 @@ auto open_state(boost::asio::any_io_executor executor, boost::asio::ip::udp::end
 }
 
 auto adopt_state(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::bool_t server) -> std::expected<std::shared_ptr<reliable_channel_state>, modulus::error> {
-  auto valid = validate_config(config);
+  auto valid = ligo::detail::validate_config(config);
   if (!valid) {
     return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(valid.error());
   }
-  auto dtls = make_dtls_state(security, server);
+  auto dtls = ligo::detail::make_dtls_state(security, server);
   if (!dtls) {
     return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(dtls.error());
   }
@@ -673,7 +360,7 @@ auto send_payload_on_strand(std::shared_ptr<reliable_channel_state> state, boost
     co_return ligo::detail::fail_void(ligo::detail::unavailable("too many UDP messages in flight"));
   }
   auto const sequence = state->next_send_sequence;
-  auto body = make_data(state->session_id, sequence, std::move(payload));
+  auto body = ligo::detail::make_data(state->session_id, sequence, std::move(payload));
   if (body.size() > max_datagram_size) {
     co_return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP payload is too large"));
   }
@@ -691,7 +378,7 @@ auto send_payload_on_strand(std::shared_ptr<reliable_channel_state> state, boost
       finish_send(state, sequence);
       co_return ligo::detail::fail_void(ligo::detail::unavailable("UDP channel closed"));
     }
-    auto datagrams = protect_dtls_payload(*state, body);
+    auto datagrams = ligo::detail::protect_dtls_payload(state->dtls.get(), body);
     if (!datagrams) {
       finish_send(state, sequence);
       co_return ligo::detail::fail_void(datagrams.error());
@@ -899,19 +586,7 @@ auto endpoint_from_config(ligo::endpoint_config const &config) -> std::expected<
   });
 }
 
-auto ssl_ctx_deleter::operator()(SSL_CTX *context) const noexcept -> void {
-  SSL_CTX_free(context);
-}
-
-auto ssl_deleter::operator()(SSL *ssl) const noexcept -> void {
-  SSL_free(ssl);
-}
-
-auto bio_deleter::operator()(BIO *bio) const noexcept -> void {
-  BIO_free(bio);
-}
-
-reliable_channel_state::reliable_channel_state(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint local_endpoint, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::unique_ptr<dtls_state> dtls)
+reliable_channel_state::reliable_channel_state(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint local_endpoint, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::unique_ptr<ligo::detail::dtls_state> dtls)
     : socket{std::move(socket)}, local_endpoint{std::move(local_endpoint)}, peer{std::move(peer)}, config{config}, security{std::move(security)}, dtls{std::move(dtls)}, session_id{session_id} {
   notify_timer.expires_at(std::chrono::steady_clock::time_point::max());
   close_timer.expires_at(std::chrono::steady_clock::time_point::max());

tests/CMakeLists.txt

@@ -1,20 +1,21 @@
-add_executable(ligo-test
+add_executable(ligo_test
   test.cpp
+  lifecycle.cpp
 )
 find_package(Threads REQUIRED)
-target_include_directories(ligo-test PRIVATE
+target_include_directories(ligo_test PRIVATE
   ${CMAKE_CURRENT_SOURCE_DIR}
 )
-target_link_libraries(ligo-test PRIVATE
+target_link_libraries(ligo_test PRIVATE
   ligo::ligo
   Threads::Threads
 )
-target_compile_features(ligo-test PRIVATE cxx_std_23)
-target_compile_definitions(ligo-test PRIVATE
+target_compile_features(ligo_test PRIVATE cxx_std_23)
+target_compile_definitions(ligo_test PRIVATE
   $<$<CONFIG:Debug>:LIGO_EXPECT_DEBUG_ASSERTS>
   $<$<NOT:$<CONFIG:Debug>>:LIGO_EXPECT_NDEBUG>
 )
-add_test(NAME ligo COMMAND ligo-test)
+add_test(NAME ligo COMMAND ligo_test --log_level=test_suite)
 set_tests_properties(ligo PROPERTIES TIMEOUT 60)
 add_custom_target(check
   COMMAND ${CMAKE_CTEST_COMMAND}

tests/lifecycle.cpp

@@ -0,0 +1,75 @@
+#include <boost/test/unit_test.hpp>
+#include <boost/asio/co_spawn.hpp>
+#include <boost/asio/executor_work_guard.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/use_future.hpp>
+#include <ligo/detail/lifecycle.hpp>
+#include <future>
+#include <thread>
+
+
+BOOST_AUTO_TEST_CASE(session_drain_handles_notifications_before_and_during_registration) {
+  for (int iteration = 0; iteration < 64; ++iteration) {
+    auto io = boost::asio::io_context{};
+    auto registry = ligo::detail::session_registry<int>{};
+    auto session = std::make_shared<int>(1);
+    registry.add(session);
+    auto result = boost::asio::co_spawn(io, registry.wait_empty(std::chrono::seconds{2}), boost::asio::use_future);
+    auto work = boost::asio::make_work_guard(io);
+    auto first = std::thread{[&] { io.run(); }};
+    auto second = std::thread{[&] { io.run(); }};
+    registry.remove(session);
+    auto ready = result.wait_for(std::chrono::milliseconds{500}) == std::future_status::ready;
+    work.reset();
+    io.stop();
+    first.join();
+    second.join();
+    BOOST_REQUIRE(ready);
+    BOOST_CHECK(result.get());
+  }
+}
+
+BOOST_AUTO_TEST_CASE(session_drain_keeps_absolute_deadline_under_repeated_notifications) {
+  auto io = boost::asio::io_context{};
+  auto registry = ligo::detail::session_registry<int>{};
+  auto session = std::make_shared<int>(1);
+  registry.add(session);
+  auto result = boost::asio::co_spawn(io, registry.wait_empty(std::chrono::milliseconds{20}), boost::asio::use_future);
+  auto work = boost::asio::make_work_guard(io);
+  auto first = std::thread{[&] { io.run(); }};
+  auto second = std::thread{[&] { io.run(); }};
+  for (int i = 0; i < 200 && result.wait_for(std::chrono::milliseconds{1}) != std::future_status::ready; ++i) {
+    registry.notify_waiters();
+  }
+  auto ready = result.wait_for(std::chrono::milliseconds{0}) == std::future_status::ready;
+  work.reset();
+  io.stop();
+  first.join();
+  second.join();
+  BOOST_REQUIRE(ready);
+  auto value = result.get();
+  BOOST_REQUIRE(!value);
+  BOOST_CHECK(value.error().code() == ligo::detail::error_timeout);
+}
+
+BOOST_AUTO_TEST_CASE(session_drain_observes_expiry_and_removal_before_wait_is_armed) {
+  auto io = boost::asio::io_context{};
+  auto registry = ligo::detail::session_registry<int>{};
+  auto session = std::make_shared<int>(1);
+  registry.add(session);
+  auto timer = std::make_shared<boost::asio::steady_timer>(boost::asio::make_strand(io));
+  registry.waiters.push_back(timer);
+  // A notification is queued before registration. It must neither extend an
+  // elapsed deadline nor prevent observing a session that already finished.
+  registry.notify_waiters();
+  auto expired = boost::asio::co_spawn(timer->get_executor(), registry.wait_registered(timer, std::chrono::steady_clock::now()), boost::asio::use_future);
+  io.run();
+  auto result = expired.get();
+  BOOST_REQUIRE(!result);
+  BOOST_CHECK(result.error().code() == ligo::detail::error_timeout);
+  io.restart();
+  registry.remove(session);
+  auto drained = boost::asio::co_spawn(timer->get_executor(), registry.wait_registered(timer, std::chrono::steady_clock::now() + std::chrono::hours{1}), boost::asio::use_future);
+  io.run();
+  BOOST_CHECK(drained.get());
+}

tests/test.cpp

@@ -1,6 +1,7 @@
 #include <atomic>
 #include <chrono>
 #include <cstdint>
+#include <exception>
 #include <fstream>
 #include <future>
 #include <memory>
@@ -2256,3 +2257,104 @@ BOOST_AUTO_TEST_CASE(ligo_udp_multithreaded_io_context_round_trip_succeeds) {
   BOOST_REQUIRE(message);
   BOOST_CHECK(message->payload.as_object().at("ok").as_bool());
 }
+
+BOOST_AUTO_TEST_CASE(ligo_http_client_operations_capture_inputs_before_await) {
+  auto io = boost::asio::io_context{};
+  auto client = ligo::http::client{{"127.0.0.1", 1}};
+  auto request = ligo::http::request{.target = ""};
+  auto call = client.call(request);
+  auto exchange = client.exchange(request);
+  client = ligo::http::client{{"127.0.0.1", 0}};
+  request.target = "/";
+  auto called = boost::asio::co_spawn(io, std::move(call), boost::asio::use_future);
+  auto exchanged = boost::asio::co_spawn(io, std::move(exchange), boost::asio::use_future);
+  io.run();
+  auto call_result = called.get();
+  BOOST_REQUIRE(!call_result);
+  BOOST_CHECK_EQUAL(call_result.error().code(), ligo::detail::error_invalid_argument);
+  BOOST_CHECK_EQUAL(call_result.error().message(), "HTTP target must not be empty");
+  auto exchange_result = exchanged.get();
+  BOOST_REQUIRE(!exchange_result);
+  BOOST_CHECK_EQUAL(exchange_result.error().code(), ligo::detail::error_invalid_argument);
+  BOOST_CHECK_EQUAL(exchange_result.error().message(), "HTTP target must not be empty");
+}
+
+BOOST_AUTO_TEST_CASE(ligo_http_client_call_outlives_client_and_request) {
+  auto io = boost::asio::io_context{};
+  // Both the helper's client and this temporary request expire before io.run().
+  auto future = boost::asio::co_spawn(io, http_client_call("127.0.0.1", 1, {.target = ""}), boost::asio::use_future);
+  io.run();
+  auto result = future.get();
+  BOOST_REQUIRE(!result);
+  BOOST_CHECK_EQUAL(result.error().code(), ligo::detail::error_invalid_argument);
+  BOOST_CHECK_EQUAL(result.error().message(), "HTTP target must not be empty");
+}
+
+BOOST_AUTO_TEST_CASE(ligo_response_metadata_is_preserved_by_router) {
+  auto io = boost::asio::io_context{};
+  auto router = ligo::service_router{};
+  router.add_response_route("GET", "/metadata", [](ligo::request_context context, boost::json::value) -> boost::asio::awaitable<std::expected<ligo::service_response, modulus::error>> {
+    BOOST_CHECK_EQUAL(context.headers.at("cookie"), "session=example");
+    BOOST_CHECK_EQUAL(context.query.at("search"), "two words");
+    auto response = ligo::text_response("text/plain", "accepted");
+    response.status = 201;
+    response.headers.emplace_back("Set-Cookie", "session=replacement; HttpOnly");
+    response.headers.emplace_back("Cache-Control", "no-store");
+    co_return response;
+  });
+  auto future = boost::asio::co_spawn(io, router.call_response({.method = "GET", .route = "/metadata", .headers = {{"cookie", "session=example"}}, .query = {{"search", "two words"}}}, {}), boost::asio::use_future);
+  io.run();
+  auto result = future.get();
+  BOOST_REQUIRE(result);
+  BOOST_CHECK_EQUAL(result->status, 201);
+  BOOST_CHECK_EQUAL(result->body, "accepted");
+  BOOST_CHECK_EQUAL(result->headers.size(), 2);
+  // The old aggregate initialization remains valid.
+  auto options = ligo::http::server_options{std::chrono::milliseconds{1000}, 1024, {}, ligo::security_config{}};
+  BOOST_CHECK(!options.parse_query);
+}
+
+BOOST_AUTO_TEST_CASE(ligo_http_raw_exchange_headers_query_and_status) {
+  auto io = boost::asio::io_context{};
+  auto router = ligo::service_router{};
+  router.add_response_route("GET", "/raw", [](ligo::request_context context, boost::json::value) -> boost::asio::awaitable<std::expected<ligo::service_response, modulus::error>> {
+    BOOST_CHECK_EQUAL(context.headers.at("if-none-match"), "revision-one");
+    BOOST_CHECK_EQUAL(context.query.at("search"), "two words+");
+    auto response = ligo::text_response("text/plain", "not a JSON envelope");
+    response.status = 202;
+    response.headers = {{"ETag", "revision-two"}, {"Set-Cookie", "sample=value; HttpOnly"}};
+    co_return response;
+  });
+  auto options = ligo::http::server_options{};
+  options.parse_query = true;
+  auto server = ligo::http::server{io, {"127.0.0.1", 0}, std::move(router), options};
+  BOOST_REQUIRE(server.start());
+  auto exercise = [&]() -> boost::asio::awaitable<void> {
+    auto client = ligo::http::client{{"127.0.0.1", server.selected_port()}};
+    // GCC 12 can ICE on initializer-list temporaries in co_await expressions.
+    auto request = ligo::http::request{.method = ligo::http::method::GET, .target = "/raw?search=two+words%2B"};
+    request.headers.emplace("If-None-Match", "revision-one");
+    auto result = co_await client.exchange(request);
+    BOOST_REQUIRE(result);
+    BOOST_CHECK_EQUAL(result->status, 202);
+    BOOST_CHECK_EQUAL(result->body, "not a JSON envelope");
+    BOOST_CHECK_EQUAL(result->headers.at("etag"), "revision-two");
+    auto invalid_request = ligo::http::request{.method = ligo::http::method::GET, .target = "/raw?search=%ZZ"};
+    auto invalid = co_await client.exchange(invalid_request);
+    BOOST_REQUIRE(invalid);
+    BOOST_CHECK_EQUAL(invalid->status, 400);
+    auto duplicate_request = ligo::http::request{.method = ligo::http::method::GET, .target = "/raw?x=1&x=2"};
+    auto duplicate = co_await client.exchange(duplicate_request);
+    BOOST_REQUIRE(duplicate);
+    BOOST_CHECK_EQUAL(duplicate->status, 400);
+  };
+  auto failure = std::exception_ptr{};
+  boost::asio::co_spawn(io, exercise(), [&](std::exception_ptr error) {
+    failure = error;
+    server.stop();
+  });
+  io.run();
+  if (failure) {
+    std::rethrow_exception(failure);
+  }
+}