Cleanup

.gitea/workflows/ci.yml

@@ -47,7 +47,7 @@ jobs:
           key: conan-${{ runner.os }}-${{ hashFiles('conanfile.py', 'modulus.json') }}
           restore-keys: conan-${{ runner.os }}-
       - name: Configure native release
-        run: python3 modulus.py configure --target x86_64_gcc12_release
+        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
@@ -63,4 +63,4 @@ jobs:
           CONAN_TOKEN: ${{ secrets.CONAN_TOKEN }}
           LIGO_CONAN_UPLOAD: "1"
           MODULUS_ALLOW_PACKAGE_REPLACE: "1"
-        run: python3 modulus.py package --all --replace
+        run: python3 modulus.py package --all --replace --update

conanfile.py

@@ -28,8 +28,11 @@ class ligo_conan(conan.ConanFile):
     "fPIC": True,
     "with_tests": False,
   }
-  exports = "modulus.json"
+  exports = "modulus.json", "!build", "!.git", "!test_package/build"
   exports_sources = (
+    "!build",
+    "!.git",
+    "!test_package/build",
     "CMakeLists.txt",
     "cmake/*.in",
     "include/*.hpp",
@@ -37,6 +40,7 @@ class ligo_conan(conan.ConanFile):
     "include/ligo/detail/*.hpp",
     "src/*.cpp",
     "src/detail/*.cpp",
+    "src/detail/*.hpp",
     "tests/CMakeLists.txt",
     "tests/*.cpp",
     "tests/*.hpp",

include/ligo/detail/lifecycle.hpp

@@ -202,6 +202,9 @@ auto ligo::detail::stop_accepting_server(std::shared_ptr<S> state) -> void {
     return;
   }
   state->sessions->cancel_all();
+  if constexpr (requires { state->tls_sessions; }) {
+    state->tls_sessions->cancel_all();
+  }
   auto server_executor = state->strand;
   boost::asio::post(server_executor, [state = std::move(state)] {
     auto ignored = boost::system::error_code{};
@@ -216,6 +219,9 @@ auto ligo::detail::stop_accepting_server_now(std::shared_ptr<S> const &state) ->
     return;
   }
   state->sessions->cancel_all();
+  if constexpr (requires { state->tls_sessions; }) {
+    state->tls_sessions->cancel_all();
+  }
   auto ignored = boost::system::error_code{};
   state->acceptor->close(ignored);
 }
@@ -232,6 +238,9 @@ auto ligo::detail::async_stop_accepting_server(std::shared_ptr<S> state, std::ch
     should_stop = !state->stopped.exchange(true);
     if (should_stop) {
       state->sessions->cancel_all();
+      if constexpr (requires { state->tls_sessions; }) {
+        state->tls_sessions->cancel_all();
+      }
     }
   }
   if (should_stop) {
@@ -239,5 +248,12 @@ auto ligo::detail::async_stop_accepting_server(std::shared_ptr<S> state, std::ch
     auto ignored = boost::system::error_code{};
     state->acceptor->close(ignored);
   }
-  co_return co_await state->sessions->wait_empty(timeout);
+  auto drained = co_await state->sessions->wait_empty(timeout);
+  if (!drained) {
+    co_return drained;
+  }
+  if constexpr (requires { state->tls_sessions; }) {
+    co_return co_await state->tls_sessions->wait_empty(timeout);
+  }
+  co_return drained;
 }

modulus.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+# Canonical consumer bootstrap. Copy this file unchanged to a project's modulus.py.
 import json
 import os
 import pathlib
@@ -17,6 +18,7 @@ COMMANDS = (
   "lsp",
   "clangd",
   "format",
+  "run",
 )
 COMMON_DIR = pathlib.Path("build") / "build_common"
 TOOLS_DIR = pathlib.Path("build") / "build_tools"
@@ -66,13 +68,6 @@ def runtime_ready(root):
     and (root / RUNTIME_DIR / "modulus" / "format_layout.py").is_file()
   )
 
-def optional_credentials():
-  user = os.environ.get("CONAN_USER")
-  token = os.environ.get("CONAN_TOKEN")
-  if bool(user) != bool(token):
-    raise SystemExit("CONAN_USER and CONAN_TOKEN must either both be set or both be unset")
-  return user, token
-
 def env_flag(name):
   return os.environ.get(name, "").lower() in {"1", "true", "yes", "on"}
 
@@ -128,7 +123,7 @@ RUN python -m pip install --no-cache-dir conan=={CONAN_VERSION}
 WORKDIR /workspace
 """
 
-def ensure_fetch_image():
+def ensure_fetch_image(offline=False):
   inspected = subprocess.run(
     [
       "docker",
@@ -140,12 +135,15 @@ def ensure_fetch_image():
     stderr=subprocess.DEVNULL,
     check=False,
   )
-  if inspected.returncode == 0:
+  if inspected.returncode == 0 and (offline or not docker_platform_args()):
     return
+  if offline:
+    raise SystemExit('offline bootstrap requires the existing Modulus fetch image or MODULUS_DIRECT=1 with Conan installed')
   run(
     [
       "docker",
       "build",
+      *docker_platform_args(),
       "--tag",
       FETCH_IMAGE,
       "-",
@@ -156,6 +154,10 @@ def ensure_fetch_image():
 def in_container():
   return pathlib.Path("/.dockerenv").exists()
 
+def docker_platform_args():
+  platform = os.environ.get("MODULUS_DOCKER_PLATFORM", "").strip()
+  return ["--platform", platform] if platform else []
+
 def docker_workspace_args(root):
   if in_container():
     container = os.environ.get("HOSTNAME", "").strip()
@@ -172,18 +174,24 @@ def docker_workspace_args(root):
 
 def fetch_env(root):
   env = os.environ.copy()
+  for name in list(env):
+    if name in {"CONAN_USER", "CONAN_TOKEN"} or name.startswith(("CONAN_LOGIN_USERNAME", "CONAN_PASSWORD")):
+      del env[name]
   env["CONAN_HOME"] = str(root / CONAN_HOME_DIR)
+  env["CONAN_NON_INTERACTIVE"] = "1"
   return env
 
-def run_fetch(root, command, capture=False, display_command=None, check=True):
+def run_fetch(root, command, capture=False, display_command=None, check=True, offline=False):
   if direct_mode(root):
     return run(command, capture=capture, display_command=display_command, env=fetch_env(root), check=check)
-  ensure_fetch_image()
+  ensure_fetch_image(offline=offline)
   docker_command = [
     "docker",
     "run",
     "--rm",
     "--interactive",
+    *(["--pull=never", "--network=none"] if offline else []),
+    *docker_platform_args(),
     "--user",
     f"{os.getuid()}:{os.getgid()}",
     "--env",
@@ -191,9 +199,7 @@ def run_fetch(root, command, capture=False, display_command=None, check=True):
     "--env",
     "CONAN_HOME",
     "--env",
-    "CONAN_USER",
-    "--env",
-    "CONAN_TOKEN",
+    "CONAN_NON_INTERACTIVE",
     *docker_workspace_args(root),
     "--workdir",
     str(root),
@@ -203,7 +209,6 @@ def run_fetch(root, command, capture=False, display_command=None, check=True):
   return run(docker_command, capture=capture, display_command=display_command, env=fetch_env(root), check=check)
 
 def configure_remote(root, remote_name, remote_url):
-  user, token = optional_credentials()
   run_fetch(
     root,
     [
@@ -215,29 +220,7 @@ def configure_remote(root, remote_name, remote_url):
       remote_url,
     ],
   )
-  if not user and not token:
-    return
-  run_fetch(
-    root,
-    [
-      "conan",
-      "remote",
-      "login",
-      remote_name,
-      user,
-      "-p",
-      token,
-    ],
-    display_command=[
-      "conan",
-      "remote",
-      "login",
-      remote_name,
-      user,
-      "-p",
-      "<hidden>",
-    ],
-  )
+  run_fetch(root, ["conan", "remote", "logout", remote_name])
 
 def package_refs(data):
   refs = []
@@ -274,46 +257,26 @@ def package_refs(data):
             refs.append(f"{PACKAGE_REF}#{recipe_revision}:{package_id}")
   return sorted(refs)
 
-def remove_cached_modulus(root):
-  run_fetch(
-    root,
-    [
-      "conan",
-      "remove",
-      PACKAGE_REF + ":*",
-      "-c",
-    ],
-    check=False,
-  )
-
-def download_modulus(root, remote_name):
-  remove_cached_modulus(root)
-  run_fetch(
-    root,
-    [
-      "conan",
-      "download",
-      PACKAGE_REF + ":*",
-      "-r",
-      remote_name,
-    ],
-  )
+def find_modulus(root, remote_name=None):
   completed = run_fetch(
     root,
     [
       "conan",
       "list",
-      PACKAGE_REF + ":*",
-      "-c",
+      PACKAGE_REF + "#latest:*#latest",
+      *(["-r", remote_name] if remote_name else ["-c"]),
       "-f",
       "json",
     ],
     capture=True,
+    offline=remote_name is None,
   )
   refs = package_refs(json.loads(completed.stdout))
   if not refs:
-    raise SystemExit(f"no downloaded {PACKAGE_REF} package found in the Conan cache")
+    raise SystemExit(f"no {PACKAGE_REF} package available in {remote_name or 'the local Conan cache'}")
   package_ref = refs[-1]
+  if remote_name:
+    run_fetch(root, ["conan", "download", package_ref, "-r", remote_name])
   completed = run_fetch(
     root,
     [
@@ -323,24 +286,25 @@ def download_modulus(root, remote_name):
       package_ref,
     ],
     capture=True,
+    offline=remote_name is None,
   )
   package_root = pathlib.Path(completed.stdout.strip())
   if not (package_root / "modulus" / "export.py").is_file() or not (package_root / "tools").is_dir():
     raise SystemExit(f"downloaded {PACKAGE_REF} package does not contain Modulus export files: {package_root}")
   return package_root
 
-def materialize_runtime(root):
+def materialize_runtime(root, no_remote=False):
   remote_name, remote_url = remote_config(root)
-  configure_remote(root, remote_name, remote_url)
-  package_root = download_modulus(root, remote_name)
+  if not no_remote:
+    configure_remote(root, remote_name, remote_url)
+  package_root = find_modulus(root, None if no_remote else remote_name)
   sys.path.insert(0, str(package_root))
   import modulus.export
   modulus.export.sync_project(root, package_root)
 
-def ensure_runtime(root):
-  if runtime_ready(root):
-    return
-  materialize_runtime(root)
+def ensure_runtime(root, update=False, no_remote=False):
+  if update or not runtime_ready(root):
+    materialize_runtime(root, no_remote=no_remote)
 
 def delegate(command, argv):
   root = repo_root()
@@ -349,7 +313,7 @@ def delegate(command, argv):
       print(f"Modulus {command} tooling is missing; run ./modulus.py configure first", file=sys.stderr)
       return 1
   else:
-    ensure_runtime(root)
+    ensure_runtime(root, update=command == "configure" and "--update" in argv, no_remote="--no-remote" in argv)
   sys.path.insert(0, str(root / RUNTIME_DIR))
   import modulus.cli
   previous_argv = sys.argv

src/detail/stream_server.hpp

@@ -0,0 +1,85 @@
+#pragma once
+
+#include <cassert>
+#include <string_view>
+#include <utility>
+
+#include <boost/asio/co_spawn.hpp>
+#include <boost/asio/detached.hpp>
+#include <boost/asio/strand.hpp>
+
+#include <ligo/detail/lifecycle.hpp>
+#include <ligo/security.hpp>
+#include <ligo/service.hpp>
+
+
+namespace ligo::detail {
+
+struct stream_listener {
+  stream_listener(boost::asio::any_io_executor executor, ligo::service_router router)
+      : executor{std::move(executor)}, strand{boost::asio::make_strand(this->executor)}, acceptor{std::make_shared<boost::asio::ip::tcp::acceptor>(strand)}, router{std::make_shared<ligo::service_router>(std::move(router))} {}
+  boost::asio::any_io_executor executor;
+  boost::asio::strand<boost::asio::any_io_executor> strand;
+  std::shared_ptr<boost::asio::ip::tcp::acceptor> acceptor{};
+  std::shared_ptr<ligo::service_router const> router{};
+  std::shared_ptr<ligo::detail::session_registry<boost::beast::tcp_stream>> sessions{std::make_shared<ligo::detail::session_registry<boost::beast::tcp_stream>>()};
+  std::shared_ptr<ligo::detail::session_registry<boost::asio::ssl::stream<boost::asio::ip::tcp::socket>>> tls_sessions{std::make_shared<ligo::detail::session_registry<boost::asio::ssl::stream<boost::asio::ip::tcp::socket>>>()};
+  std::shared_ptr<boost::asio::ssl::context> tls_context{};
+  std::atomic_uint16_t selected_port{};
+  std::atomic<std::bool_t> start_called{false};
+  std::atomic<std::bool_t> stopped{false};
+  mutable std::mutex lifecycle_mutex{};
+};
+
+template <typename STATE, typename ACCEPT, typename ACCEPT_TLS>
+auto start_stream_server(std::shared_ptr<STATE> const &state, std::string_view protocol, ACCEPT accept_loop, ACCEPT_TLS accept_tls_loop) -> std::expected<void, modulus::error>;
+
+}
+
+template <typename STATE, typename ACCEPT, typename ACCEPT_TLS>
+auto ligo::detail::start_stream_server(std::shared_ptr<STATE> const &state, std::string_view protocol, ACCEPT accept_loop, ACCEPT_TLS accept_tls_loop) -> std::expected<void, modulus::error> {
+  auto lock = std::scoped_lock{state->lifecycle_mutex};
+  auto valid_timeout = ligo::detail::validate_positive_timeout(state->options.timeout, std::string{protocol} + " server timeout");
+  if (!valid_timeout) {
+    return valid_timeout;
+  }
+  if (state->options.security.mode == ligo::security_mode::DTLS) {
+    return ligo::detail::fail_void(ligo::unsupported_security(std::string{protocol} + " server", state->options.security.mode));
+  }
+  if (state->start_called.exchange(true)) {
+    return ligo::detail::fail_void(ligo::detail::unavailable(std::string{protocol} + " server has already been started"));
+  }
+  if (state->options.security.mode == ligo::security_mode::TLS) {
+    auto valid_config = ligo::validate_tls_server_config(protocol, state->options.security);
+    if (!valid_config) {
+      state->start_called.store(false);
+      return valid_config;
+    }
+    auto context = ligo::make_tls_server_context(state->options.security);
+    if (!context) {
+      state->start_called.store(false);
+      return ligo::detail::fail_void(context.error());
+    }
+    auto valid_context = ligo::validate_tls_server_context(*context);
+    if (!valid_context) {
+      state->start_called.store(false);
+      return valid_context;
+    }
+    state->tls_context = std::move(*context);
+  }
+  auto selected = ligo::detail::start_tcp_acceptor(*state->acceptor, state->endpoint.host, state->endpoint.port);
+  if (!selected) {
+    state->start_called.store(false);
+    return ligo::detail::fail_void(selected.error());
+  }
+  state->selected_port.store(*selected);
+  assert(state->selected_port.load() == *selected);
+  state->stopped.store(false);
+  modulus::log_debug("ligo " + std::string{protocol} + " server started on port " + std::to_string(*selected));
+  if (state->options.security.mode == ligo::security_mode::TLS) {
+    boost::asio::co_spawn(state->strand, accept_tls_loop(state), boost::asio::detached);
+  } else {
+    boost::asio::co_spawn(state->strand, accept_loop(state), boost::asio::detached);
+  }
+  return {};
+}

src/http.cpp

@@ -36,7 +36,7 @@
 #include <modulus.hpp>
 
 #include <ligo/detail/errors.hpp>
-#include <ligo/detail/lifecycle.hpp>
+#include "detail/stream_server.hpp"
 #include <ligo/detail/state.hpp>
 #include <ligo/detail/transport.hpp>
 #include <ligo/config.hpp>
@@ -46,21 +46,10 @@
 
 namespace {
 
-struct http_server_state {
+struct http_server_state : ligo::detail::stream_listener {
   http_server_state(boost::asio::any_io_executor executor, ligo::http::endpoint endpoint, ligo::service_router router, ligo::http::server_options options);
-  boost::asio::any_io_executor executor;
-  boost::asio::strand<boost::asio::any_io_executor> strand;
-  std::shared_ptr<boost::asio::ip::tcp::acceptor> acceptor{};
   ligo::http::endpoint endpoint{};
   ligo::http::server_options options{};
-  std::shared_ptr<ligo::service_router const> router{};
-  std::shared_ptr<ligo::detail::session_registry<boost::beast::tcp_stream>> sessions{std::make_shared<ligo::detail::session_registry<boost::beast::tcp_stream>>()};
-  std::shared_ptr<ligo::detail::session_registry<boost::asio::ssl::stream<boost::asio::ip::tcp::socket>>> tls_sessions{std::make_shared<ligo::detail::session_registry<boost::asio::ssl::stream<boost::asio::ip::tcp::socket>>>()};
-  std::shared_ptr<boost::asio::ssl::context> tls_context{};
-  std::atomic_uint16_t selected_port{};
-  std::atomic<std::bool_t> start_called{false};
-  std::atomic<std::bool_t> stopped{false};
-  mutable std::mutex lifecycle_mutex{};
 };
 
 auto http_status(std::int32_t code) -> boost::beast::http::status {
@@ -297,111 +286,26 @@ auto uninitialized_http_server() -> std::expected<void, modulus::error> {
 }
 
 auto start_http_server(std::shared_ptr<http_server_state> const &state) -> std::expected<void, modulus::error> {
-  assert(!state || state->acceptor != nullptr);
-  assert(!state || state->router != nullptr);
   if (!state) {
     return uninitialized_http_server();
   }
-  auto lock = std::scoped_lock{state->lifecycle_mutex};
-  auto valid_timeout = ligo::detail::validate_positive_timeout(state->options.timeout, "HTTP server timeout");
-  if (!valid_timeout) {
-    return valid_timeout;
-  }
-  if (state->options.security.mode == ligo::security_mode::DTLS) {
-    return ligo::detail::fail_void(ligo::unsupported_security("HTTP server", state->options.security.mode));
-  }
-  if (state->start_called.exchange(true)) {
-    return ligo::detail::fail_void(ligo::detail::unavailable("HTTP server has already been started"));
-  }
-  if (state->options.security.mode == ligo::security_mode::TLS) {
-    auto valid_config = ligo::validate_tls_server_config("HTTP", state->options.security);
-    if (!valid_config) {
-      state->start_called.store(false);
-      return valid_config;
-    }
-    auto context = ligo::make_tls_server_context(state->options.security);
-    if (!context) {
-      state->start_called.store(false);
-      return ligo::detail::fail_void(context.error());
-    }
-    auto valid_context = ligo::validate_tls_server_context(*context);
-    if (!valid_context) {
-      state->start_called.store(false);
-      return valid_context;
-    }
-    state->tls_context = std::move(*context);
-  }
-  auto selected = ligo::detail::start_tcp_acceptor(*state->acceptor, state->endpoint.host, state->endpoint.port);
-  if (!selected) {
-    state->start_called.store(false);
-    return ligo::detail::fail_void(selected.error());
-  }
-  state->selected_port.store(*selected);
-  assert(state->selected_port.load() == *selected);
-  state->stopped.store(false);
-  modulus::log_debug("ligo HTTP server started on port " + std::to_string(*selected));
-  if (state->options.security.mode == ligo::security_mode::TLS) {
-    boost::asio::co_spawn(state->strand, accept_tls_loop(state), boost::asio::detached);
-  } else {
-    boost::asio::co_spawn(state->strand, accept_loop(state), boost::asio::detached);
-  }
-  return {};
+  return ligo::detail::start_stream_server(state, "HTTP", accept_loop, accept_tls_loop);
 }
 
 auto stop_http_server(std::shared_ptr<http_server_state> state) -> void {
-  assert(state != nullptr);
-  auto lock = std::scoped_lock{state->lifecycle_mutex};
-  if (state->stopped.exchange(true)) {
-    return;
-  }
   modulus::log_debug("ligo HTTP server stopping");
-  state->sessions->cancel_all();
-  state->tls_sessions->cancel_all();
-  auto server_executor = state->strand;
-  boost::asio::post(server_executor, [state = std::move(state)] {
-    auto ignored = boost::system::error_code{};
-    state->acceptor->close(ignored);
-  });
+  ligo::detail::stop_accepting_server(std::move(state));
 }
 
 auto stop_http_server_now(std::shared_ptr<http_server_state> const &state) -> void {
-  auto lock = std::scoped_lock{state->lifecycle_mutex};
-  if (state->stopped.exchange(true)) {
-    return;
-  }
-  state->sessions->cancel_all();
-  state->tls_sessions->cancel_all();
-  auto ignored = boost::system::error_code{};
-  state->acceptor->close(ignored);
+  ligo::detail::stop_accepting_server_now(state);
 }
 
 auto async_stop_http_server(std::shared_ptr<http_server_state> state, std::chrono::milliseconds timeout) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
   if (!state) {
     co_return uninitialized_http_server();
   }
-  auto valid_timeout = ligo::detail::validate_positive_timeout(timeout, "shutdown timeout");
-  if (!valid_timeout) {
-    co_return valid_timeout;
-  }
-  auto should_stop = false;
-  {
-    auto lock = std::scoped_lock{state->lifecycle_mutex};
-    should_stop = !state->stopped.exchange(true);
-    if (should_stop) {
-      state->sessions->cancel_all();
-      state->tls_sessions->cancel_all();
-    }
-  }
-  if (should_stop) {
-    co_await boost::asio::post(state->strand, boost::asio::use_awaitable);
-    auto ignored = boost::system::error_code{};
-    state->acceptor->close(ignored);
-  }
-  auto drained = co_await state->sessions->wait_empty(timeout);
-  if (!drained) {
-    co_return drained;
-  }
-  co_return co_await state->tls_sessions->wait_empty(timeout);
+  co_return co_await ligo::detail::async_stop_accepting_server(std::move(state), timeout);
 }
 
 auto selected_http_port(std::shared_ptr<http_server_state> const &state) -> std::uint16_t {
@@ -546,7 +450,7 @@ auto call_http_endpoint(ligo::http::endpoint endpoint, std::shared_ptr<ligo::htt
 }
 
 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)
-    : executor{std::move(executor)}, strand{boost::asio::make_strand(this->executor)}, acceptor{std::make_shared<boost::asio::ip::tcp::acceptor>(strand)}, endpoint{std::move(endpoint)}, options{options}, router{std::make_shared<ligo::service_router>(std::move(router))} {}
+    : ligo::detail::stream_listener{std::move(executor), std::move(router)}, endpoint{std::move(endpoint)}, options{options} {}
 
 }
 

src/tcp.cpp

@@ -39,7 +39,7 @@
 #include <modulus.hpp>
 
 #include <ligo/detail/errors.hpp>
-#include <ligo/detail/lifecycle.hpp>
+#include "detail/stream_server.hpp"
 #include <ligo/detail/state.hpp>
 #include <ligo/detail/transport.hpp>
 #include <ligo/config.hpp>
@@ -49,21 +49,10 @@
 
 namespace {
 
-struct tcp_server_state {
+struct tcp_server_state : ligo::detail::stream_listener {
   tcp_server_state(boost::asio::any_io_executor executor, ligo::tcp::endpoint endpoint, ligo::service_router router, ligo::tcp::server_options options);
-  boost::asio::any_io_executor executor;
-  boost::asio::strand<boost::asio::any_io_executor> strand;
-  std::shared_ptr<boost::asio::ip::tcp::acceptor> acceptor{};
   ligo::tcp::endpoint endpoint{};
   ligo::tcp::server_options options{};
-  std::shared_ptr<ligo::service_router const> router{};
-  std::shared_ptr<ligo::detail::session_registry<boost::beast::tcp_stream>> sessions{std::make_shared<ligo::detail::session_registry<boost::beast::tcp_stream>>()};
-  std::shared_ptr<ligo::detail::session_registry<boost::asio::ssl::stream<boost::asio::ip::tcp::socket>>> tls_sessions{std::make_shared<ligo::detail::session_registry<boost::asio::ssl::stream<boost::asio::ip::tcp::socket>>>()};
-  std::shared_ptr<boost::asio::ssl::context> tls_context{};
-  std::atomic_uint16_t selected_port{};
-  std::atomic<std::bool_t> start_called{false};
-  std::atomic<std::bool_t> stopped{false};
-  mutable std::mutex lifecycle_mutex{};
 };
 
 constexpr auto max_frame_size = std::uint32_t{16U * 1024U * 1024U};
@@ -283,111 +272,26 @@ auto uninitialized_tcp_server() -> std::expected<void, modulus::error> {
 }
 
 auto start_tcp_server(std::shared_ptr<tcp_server_state> const &state) -> std::expected<void, modulus::error> {
-  assert(!state || state->acceptor != nullptr);
-  assert(!state || state->router != nullptr);
   if (!state) {
     return uninitialized_tcp_server();
   }
-  auto lock = std::scoped_lock{state->lifecycle_mutex};
-  auto valid_timeout = ligo::detail::validate_positive_timeout(state->options.timeout, "TCP server timeout");
-  if (!valid_timeout) {
-    return valid_timeout;
-  }
-  if (state->options.security.mode == ligo::security_mode::DTLS) {
-    return ligo::detail::fail_void(ligo::unsupported_security("TCP server", state->options.security.mode));
-  }
-  if (state->start_called.exchange(true)) {
-    return ligo::detail::fail_void(ligo::detail::unavailable("TCP server has already been started"));
-  }
-  if (state->options.security.mode == ligo::security_mode::TLS) {
-    auto valid_config = ligo::validate_tls_server_config("TCP", state->options.security);
-    if (!valid_config) {
-      state->start_called.store(false);
-      return valid_config;
-    }
-    auto context = ligo::make_tls_server_context(state->options.security);
-    if (!context) {
-      state->start_called.store(false);
-      return ligo::detail::fail_void(context.error());
-    }
-    auto valid_context = ligo::validate_tls_server_context(*context);
-    if (!valid_context) {
-      state->start_called.store(false);
-      return valid_context;
-    }
-    state->tls_context = std::move(*context);
-  }
-  auto selected = ligo::detail::start_tcp_acceptor(*state->acceptor, state->endpoint.host, state->endpoint.port);
-  if (!selected) {
-    state->start_called.store(false);
-    return ligo::detail::fail_void(selected.error());
-  }
-  state->selected_port.store(*selected);
-  assert(state->selected_port.load() == *selected);
-  state->stopped.store(false);
-  modulus::log_debug("ligo TCP server started on port " + std::to_string(*selected));
-  if (state->options.security.mode == ligo::security_mode::TLS) {
-    boost::asio::co_spawn(state->strand, accept_tls_loop(state), boost::asio::detached);
-  } else {
-    boost::asio::co_spawn(state->strand, accept_loop(state), boost::asio::detached);
-  }
-  return {};
+  return ligo::detail::start_stream_server(state, "TCP", accept_loop, accept_tls_loop);
 }
 
 auto stop_tcp_server(std::shared_ptr<tcp_server_state> state) -> void {
-  assert(state != nullptr);
-  auto lock = std::scoped_lock{state->lifecycle_mutex};
-  if (state->stopped.exchange(true)) {
-    return;
-  }
   modulus::log_debug("ligo TCP server stopping");
-  state->sessions->cancel_all();
-  state->tls_sessions->cancel_all();
-  auto server_executor = state->strand;
-  boost::asio::post(server_executor, [state = std::move(state)] {
-    auto ignored = boost::system::error_code{};
-    state->acceptor->close(ignored);
-  });
+  ligo::detail::stop_accepting_server(std::move(state));
 }
 
 auto stop_tcp_server_now(std::shared_ptr<tcp_server_state> const &state) -> void {
-  auto lock = std::scoped_lock{state->lifecycle_mutex};
-  if (state->stopped.exchange(true)) {
-    return;
-  }
-  state->sessions->cancel_all();
-  state->tls_sessions->cancel_all();
-  auto ignored = boost::system::error_code{};
-  state->acceptor->close(ignored);
+  ligo::detail::stop_accepting_server_now(state);
 }
 
 auto async_stop_tcp_server(std::shared_ptr<tcp_server_state> state, std::chrono::milliseconds timeout) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
   if (!state) {
     co_return uninitialized_tcp_server();
   }
-  auto valid_timeout = ligo::detail::validate_positive_timeout(timeout, "shutdown timeout");
-  if (!valid_timeout) {
-    co_return valid_timeout;
-  }
-  auto should_stop = false;
-  {
-    auto lock = std::scoped_lock{state->lifecycle_mutex};
-    should_stop = !state->stopped.exchange(true);
-    if (should_stop) {
-      state->sessions->cancel_all();
-      state->tls_sessions->cancel_all();
-    }
-  }
-  if (should_stop) {
-    co_await boost::asio::post(state->strand, boost::asio::use_awaitable);
-    auto ignored = boost::system::error_code{};
-    state->acceptor->close(ignored);
-  }
-  auto drained = co_await state->sessions->wait_empty(timeout);
-  if (!drained) {
-    co_return drained;
-  }
-  co_return co_await state->tls_sessions->wait_empty(timeout);
+  co_return co_await ligo::detail::async_stop_accepting_server(std::move(state), timeout);
 }
 
 auto selected_tcp_port(std::shared_ptr<tcp_server_state> const &state) -> std::uint16_t {
@@ -498,7 +402,7 @@ auto validate_tls_connect_security(ligo::security_config const &security) -> std
 }
 
 tcp_server_state::tcp_server_state(boost::asio::any_io_executor executor, ligo::tcp::endpoint endpoint, ligo::service_router router, ligo::tcp::server_options options)
-    : executor{std::move(executor)}, strand{boost::asio::make_strand(this->executor)}, acceptor{std::make_shared<boost::asio::ip::tcp::acceptor>(strand)}, endpoint{std::move(endpoint)}, options{options}, router{std::make_shared<ligo::service_router>(std::move(router))} {}
+    : ligo::detail::stream_listener{std::move(executor), std::move(router)}, endpoint{std::move(endpoint)}, options{options} {}
 
 }
 
@@ -749,51 +653,17 @@ auto ligo::tcp::connect_tls_configured(boost::asio::any_io_executor executor, li
 }
 
 auto ligo::tcp::connect_tls_handshaken(boost::asio::any_io_executor executor, ligo::tcp::endpoint endpoint, ligo::tcp::client_options options, ligo::security_config security) -> boost::asio::awaitable<std::expected<ligo::tcp::tls_socket, modulus::error>> {
-  auto host = endpoint.host;
-  auto valid_timeout = ligo::detail::validate_positive_timeout(options.timeout, "TCP timeout");
-  if (!valid_timeout) {
-    co_return ligo::detail::fail<ligo::tcp::tls_socket>(valid_timeout.error());
+  auto socket = co_await ligo::tcp::connect_tls_configured(std::move(executor), std::move(endpoint), options, std::move(security));
+  if (!socket) {
+    co_return ligo::detail::fail<ligo::tcp::tls_socket>(socket.error());
   }
-  auto valid_security = validate_tls_connect_security(security);
-  if (!valid_security) {
-    co_return ligo::detail::fail<ligo::tcp::tls_socket>(valid_security.error());
+  auto stream = socket->stream();
+  if (!stream) {
+    co_return ligo::detail::fail<ligo::tcp::tls_socket>(stream.error());
   }
-  if (endpoint.port == 0) {
-    co_return ligo::detail::fail<ligo::tcp::tls_socket>(ligo::detail::error_invalid_argument, "TCP port must be non-zero");
-  }
-  auto context = ligo::make_tls_client_context(security);
-  if (!context) {
-    co_return ligo::detail::fail<ligo::tcp::tls_socket>(context.error());
-  }
-  auto endpoints = co_await ligo::detail::resolve_tcp_endpoints(executor, endpoint.host, endpoint.port, options.timeout, "TCP connect timed out");
-  if (!endpoints) {
-    co_return ligo::detail::fail<ligo::tcp::tls_socket>(endpoints.error());
-  }
-  auto stream = boost::beast::tcp_stream{executor};
-  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, "TCP connect timed out");
-    co_return ligo::detail::fail<ligo::tcp::tls_socket>(failure_status);
-  }
-  auto socket = ligo::tcp::tls_socket{
-    std::move(*context),
-    stream.release_socket(),
-  };
-  auto socket_stream = socket.stream();
-  if (!socket_stream) {
-    co_return ligo::detail::fail<ligo::tcp::tls_socket>(socket_stream.error());
-  }
-  auto verification = ligo::apply_tls_client_verification(socket_stream->get(), host, security);
-  if (!verification) {
-    co_return ligo::detail::fail<ligo::tcp::tls_socket>(verification.error());
-  }
-  auto handshake = co_await ligo::detail::async_tls_handshake(socket_stream->get(), boost::asio::ssl::stream_base::client, options.timeout, "TCP TLS handshake timed out");
+  auto handshake = co_await ligo::detail::async_tls_handshake(stream->get(), boost::asio::ssl::stream_base::client, options.timeout, "TCP TLS handshake timed out");
   if (!handshake) {
     co_return ligo::detail::fail<ligo::tcp::tls_socket>(handshake.error());
   }
-  co_return std::expected<ligo::tcp::tls_socket, modulus::error>{
-    std::in_place,
-    std::move(socket),
-  };
+  co_return std::move(socket);
 }