Cleanup

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>

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.

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 timer = std::make_shared<boost::asio::steady_timer>(boost::asio::make_strand(executor));
-  auto deadline_timer = std::make_shared<boost::asio::steady_timer>(executor);
+  auto deadline = std::chrono::steady_clock::now() + timeout;
-  auto timed_out = std::make_shared<std::atomic<std::bool_t>>(false);
   {
     auto lock = std::scoped_lock{mutex};
-    waiters.push_back(notify_timer);
+    waiters.push_back(timer);
   }
-  deadline_timer->expires_after(timeout);
+  co_return co_await boost::asio::co_spawn(timer->get_executor(), wait_registered(timer, deadline), boost::asio::use_awaitable);
-  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();
 }
-  });
+
+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());
+    timer->expires_at(deadline);
-    auto [error] = co_await notify_timer->async_wait(boost::asio::as_tuple(boost::asio::use_awaitable));
+    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));
     }

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/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 ack_body = ligo::detail::make_ack(state->session_id, sequence);
-  auto datagrams = protect_dtls_payload(*state, ack_body);
+  auto datagrams = ligo::detail::protect_dtls_payload(state->dtls.get(), ack_body);
   if (!datagrams) {
     co_return ligo::detail::fail_void(datagrams.error());
   }
@@ -487,7 +174,7 @@ 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{
+    auto processed = ligo::detail::process_dtls_input(state->dtls.get(), std::string_view{
                                                                            state->receive_buffer.data(),
                                                                            size,
                                                                          });
@@ -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 session = ligo::detail::uint_member(object, "session");
-      auto sequence = uint_member(object, "sequence");
+      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 {
+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)
-  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)
     : 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,5 +1,6 @@
 add_executable(ligo_test
   test.cpp
+  lifecycle.cpp
 )
 find_package(Threads REQUIRED)
 target_include_directories(ligo_test PRIVATE

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());
+}