From ea8a83804ed8117fdef6f4f51dc6e40c6d74f4ce Mon Sep 17 00:00:00 2001 From: lojack5 <1458329+lojack5@users.noreply.github.com> Date: Sun, 4 Oct 2026 13:16:27 -0600 Subject: [PATCH 1/3] server: add new exception types for 400 responses catch all on RequestError converts to HTTP 400, can use narrower catch blocks and subclasses to refine this behavior if needed --- app/server/parallel_runtime.cpp | 4 ++-- app/server/runtime.cpp | 4 ++-- include/engine/framework/runtime/errors.h | 16 +++++++++++----- 3 files changed, 15 insertions(+), 9 deletions(-) diff --git a/app/server/parallel_runtime.cpp b/app/server/parallel_runtime.cpp index 5d7175bc7..2677fcf10 100644 --- a/app/server/parallel_runtime.cpp +++ b/app/server/parallel_runtime.cpp @@ -1283,8 +1283,8 @@ HttpResponse ParallelServerState::handle_request(const HttpRequest & request, bo else { response = error_response(404, "unknown endpoint: " + request.path, "not_found"); } - } catch (const engine::runtime::CapacityError & ex) { - // The request is too big for the device, which is the caller's to fix -- + } catch (const engine::runtime::RequestValidationError & ex) { + // Something is wrong with the request that the caller can fix -- // reporting it as 500 sends them looking for a server fault that is not // there. Checked before ServerBusyError only because both are // runtime_error; the two conditions are disjoint. diff --git a/app/server/runtime.cpp b/app/server/runtime.cpp index 9351caccc..b295b821a 100644 --- a/app/server/runtime.cpp +++ b/app/server/runtime.cpp @@ -1274,8 +1274,8 @@ HttpResponse ServerState::handle_request(const HttpRequest & request, bool use_f else { response = error_response(404, "unknown endpoint: " + request.path, "not_found"); } - } catch (const engine::runtime::CapacityError & ex) { - // The request is too big for the device, which is the caller's to fix -- + } catch (const engine::runtime::RequestValidationError & ex) { + // Something is wrong with the request that the caller can fix -- // reporting it as 500 sends them looking for a server fault that is not // there. Checked before ServerBusyError only because both are // runtime_error; the two conditions are disjoint. diff --git a/include/engine/framework/runtime/errors.h b/include/engine/framework/runtime/errors.h index fe1c3c129..7af0b7f2d 100644 --- a/include/engine/framework/runtime/errors.h +++ b/include/engine/framework/runtime/errors.h @@ -5,13 +5,19 @@ namespace engine::runtime { +// A request has a problem that the client can fix by changing request +// parameters. Distinct from a genuine internal fault: so servers should +// surface it as a client error rather than an opaque 500. +class RequestValidationError : public std::runtime_error { +public: + explicit RequestValidationError(const std::string & message) : std::runtime_error(message) {} +}; + // A request the device cannot serve AT THIS SIZE -- e.g. a transcription -// prompt plus audio whose prefill graph does not fit in VRAM. Distinct from a -// genuine internal fault: the caller can fix it by sending less, so servers -// should surface it as a client error rather than an opaque 500. -class CapacityError : public std::runtime_error { +// prompt plus audio whose prefill graph does not fit in VRAM. +class CapacityError : public RequestValidationError { public: - explicit CapacityError(const std::string & message) : std::runtime_error(message) {} + explicit CapacityError(const std::string & message) : RequestValidationError(message) {} }; } // namespace engine::runtime From c52df2a9b23001628a11b4b9eb148b9381c51ca8 Mon Sep 17 00:00:00 2001 From: lojack5 <1458329+lojack5@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:02:21 -0600 Subject: [PATCH 2/3] Test the exception handling of non-parallel ServerState::handle_request --- CMakeLists.txt | 14 ++ app/server/runtime.cpp | 2 +- app/server/runtime.h | 7 + tests/unittests/server_state_fixture.h | 181 ++++++++++++++++++ .../test_server_request_exception.cpp | 145 ++++++++++++++ 5 files changed, 348 insertions(+), 1 deletion(-) create mode 100644 tests/unittests/server_state_fixture.h create mode 100644 tests/unittests/test_server_request_exception.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 0ac72b823..6b6727a1c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -3095,6 +3095,20 @@ if (ENGINE_BUILD_TESTS OR ENGINE_BUILD_EXTENDED_TESTS OR ENGINE_BUILD_MODEL_TEST add_test(NAME server_parallel_lifecycle_test COMMAND server_parallel_lifecycle_test) set_tests_properties(server_parallel_lifecycle_test PROPERTIES TIMEOUT 90) + # Exercise the server exception conversion to HTTP responses + get_target_property(server_exception_sources audiocpp_server SOURCES) + list(FILTER server_exception_sources EXCLUDE REGEX "app/server/(main|model_installer|parallel_runtime)\\.cpp$") + add_executable(server_request_exceptions_test + tests/unittests/test_server_request_exception.cpp ${server_exception_sources}) + target_include_directories(server_request_exceptions_test PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR}/app/server "${CMAKE_CURRENT_BINARY_DIR}/generated") + target_link_libraries(server_request_exceptions_test PRIVATE engine_runtime ggml Threads::Threads) + if(WIN32) + target_link_libraries(server_request_exceptions_test PRIVATE ws2_32) + endif() + add_test(NAME server_request_exceptions_test COMMAND server_request_exceptions_test) + set_tests_properties(server_request_exceptions_test PROPERTIES TIMEOUT 60) + add_executable(session_pool_test tests/unittests/test_session_pool.cpp) target_include_directories(session_pool_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include ${CMAKE_CURRENT_SOURCE_DIR}/app/server) # TaskResult owns JSON values for custom-schema outputs. diff --git a/app/server/runtime.cpp b/app/server/runtime.cpp index b295b821a..e2c3e947f 100644 --- a/app/server/runtime.cpp +++ b/app/server/runtime.cpp @@ -1939,7 +1939,7 @@ void ServerState::ensure_model_loaded_locked(LoadedModel & model) { // That is acceptable: the 503 tells the client to retry later. } ensure_model_fits_memory(model.config); - auto registry = engine::runtime::make_default_registry(); + auto registry = registry_factory_(); engine::runtime::ModelLoadRequest load_request; load_request.model_path = model.config.path; diff --git a/app/server/runtime.h b/app/server/runtime.h index bd89aa884..77dfc86d9 100644 --- a/app/server/runtime.h +++ b/app/server/runtime.h @@ -13,6 +13,7 @@ #include "engine/framework/io/json.h" #include "engine/framework/runtime/model.h" #include "engine/framework/runtime/session.h" +#include "engine/framework/runtime/registry.h" #include #include @@ -92,6 +93,12 @@ class ServerState final : public IHttpHandler, public ServerFrontendContext { void unload(); }; + // Controlled runtime tests suppy loaders and barriers without production flags. + friend class ServerRuntimeTestAccess; + std::function registry_factory_ = [] { + return engine::runtime::make_default_registry(); + }; + // Acquire the model's run guard. `request_timeout_ms` is the caller-supplied // override, clamped by this model's configured ceiling. Throws ServerBusyError // (-> HTTP 503) once the effective timeout has elapsed. diff --git a/tests/unittests/server_state_fixture.h b/tests/unittests/server_state_fixture.h new file mode 100644 index 000000000..cc1418156 --- /dev/null +++ b/tests/unittests/server_state_fixture.h @@ -0,0 +1,181 @@ +#pragma once + +// Define Mock server, model, and model loaders to use as test fixtures +// when testing ServerState specific code that doesn't require a model +// actually loaded, nor a server started and bound to any addresses + +#include "runtime.h" +#include "engine/framework/runtime/registry.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace rt = engine::runtime; +namespace srv = minitts::server; + + +namespace { + +// lambda for test-supplied model run behavior +using RunFunction = std::function; + +inline constexpr const char * kModelFamily = "test-server-fixture"; + +// Minimal mock Offline Session +class MockSession final : public rt::IOfflineVoiceTaskSession { +public: + MockSession(const rt::TaskSpec & task, std::shared_ptr run) + : task_(task), run_(std::move(run)) {} + + std::string family() const override { return kModelFamily; } + rt::VoiceTaskKind task_kind() const override { return task_.task; } + rt::RunMode run_mode() const override { return task_.mode; } + void prepare(const rt::SessionPreparationRequest &) override {} + rt::TaskResult run(const rt::TaskRequest & request) override { return (*run_)(request); } + +private: + rt::TaskSpec task_; + std::shared_ptr run_; +}; + +// Minimal mock offline model +class MockModel final : public rt::ILoadedVoiceModel { +public: + explicit MockModel(std::shared_ptr run) + : run_(std::move(run)) { + metadata_.family = kModelFamily; + } + + const rt::ModelMetadata & metadata() const noexcept override { return metadata_; } + const rt::CapabilitySet & capabilities() const noexcept override { return capabilities_; } + std::unique_ptr create_task_session(const rt::TaskSpec & task, const rt::SessionOptions &) const { + return std::make_unique(task, run_); + } + +private: + std::shared_ptr run_; + rt::ModelMetadata metadata_; + rt::CapabilitySet capabilities_; +}; + +// Mock model loader so ServerState can "load" our mock model without any actual weights +class MockLoader final : public rt::IVoiceModelLoader { +public: + explicit MockLoader(std::shared_ptr run) : run_(std::move(run)) {} + + std::string family() const override { return kModelFamily; } + bool can_load(const rt::ModelLoadRequest &) const override { return true; } + rt::ModelInspection inspect(const rt::ModelLoadRequest &) const override { return {}; } + std::unique_ptr load(const rt::ModelLoadRequest &) const override { + return std::make_unique(run_); + } + +private: + std::shared_ptr run_; +}; +} // namespace + +// Test-only friend class of runtime::ServerState, needed +// to modify its model loader to use the mock loader. +// Declared in the minitts::server namespace originally +namespace minitts::server { + +class ServerRuntimeTestAccess { +public: + static void set_registry_factory(ServerState & state, const std::shared_ptr & loader) { + state.registry_factory_ = [loader] { + rt::ModelRegistry registry; + registry.register_loader(loader); + return registry; + }; + } + + static void add(ServerState & state, ServerModelConfig config) { + auto model = state.make_model(std::move(config)); + state.model_index_.emplace(model->config.id, state.models_.size()); + state.models_.push_back(std::move(model)); + } +}; +} // namespace minitts::server +using Access = srv::ServerRuntimeTestAccess; + +namespace { +// Test fixture: test a real ServerState object using a lazily loaded MockModel, so +// no weights are loaded, no actual server is launched. +class Fixture { +public: + explicit Fixture(RunFunction run, std::string cors = "*") + : root_(make_temp_root()), + run_(std::make_shared(std::move(run))), + loader_(std::make_shared(run_)) { + srv::ServerConfig config; + config.backend = engine::core::BackendType::Cpu; + config.ui_enabled = false; + config.ui_management = true; // needed to ensure the model loader will use our mock model + config.idle_unload_ms = 0; + config.max_loaded_models = 1; + config.min_free_memory_mb = 0; + config.cors_origins = cors; + state_ = std::make_unique(std::move(config), root_); + Access::set_registry_factory(*state_, loader_); + } + Fixture(const Fixture &) = delete; + ~Fixture() { + // Destroy ServerState before cleaning up temp directory + state_.reset(); + std::error_code ec; + std::filesystem::remove_all(root_, ec); + } + Fixture & operator= (const Fixture &) = delete; + + void add(const std::string & id, const std::string & mode = "offline") { + // ServerState's load_model needs a fake file in place when looking up model + // contracts for the fake model + std::ofstream(root_ / id) << "controlled fixture"; + // Now create and add the model, already "loaded" + srv::ServerModelConfig model; + model.id = id; + model.path = root_ / id; + model.family = loader_->family(); + model.mode = mode; + model.lazy = true; + Access::add(*state_, std::move(model)); + } + + // Test convenience methods + srv::HttpResponse handle(const srv::HttpRequest & request) { + return state_->handle(request); + } + + srv::HttpResponse post(const std::string & path, std::string body, const std::string origin = "http://server-state-test") { + srv::HttpRequest request; + request.method = "POST"; + request.path = path; + request.body = std::move(body); + request.headers["content-type"] = "application/json"; + request.headers["origin"] = origin; + return handle(request); + } + +private: + static std::filesystem::path make_temp_root() { + const auto ticks = std::chrono::steady_clock::now().time_since_epoch().count(); + auto root = std::filesystem::temp_directory_path() / + ("audio-cpp-server-state-test-" + std::to_string(ticks)); + std::filesystem::create_directories(root); + return root; + } + + std::filesystem::path root_; + std::shared_ptr run_; + std::shared_ptr loader_; + std::unique_ptr state_; +}; + +} // namespace diff --git a/tests/unittests/test_server_request_exception.cpp b/tests/unittests/test_server_request_exception.cpp new file mode 100644 index 000000000..ddf1d699c --- /dev/null +++ b/tests/unittests/test_server_request_exception.cpp @@ -0,0 +1,145 @@ +#include "server_state_fixture.h" +#include "engine/framework/runtime/errors.h" + +#include +#include +#include + + +namespace { + +void require(bool condition, const std::string & message) { + if (!condition) + throw std::runtime_error(message); +} + + +void test_capacity_error() { + // Test that CapacityError gets converted to a HTTP 400 + Fixture f( + [](const rt::TaskRequest &) -> rt::TaskResult { + throw rt::CapacityError("not enough capacity"); + } + ); + f.add("test-capacity"); + const auto response = f.post("/v1/tasks/run", + R"({"model":"test-capacity", "text":"hello"})"); + require(response.status == 400, "Expected HTTP 400, got " + std::to_string(response.status) + ": " + response.body); + require(response.body.find("not enough capacity") != std::string::npos, "HTTP response did not contain the expected error: " + response.body); + require(response.body.find("invalid_request_error") != std::string::npos, "HTTP response did not identify invalid_request_error: " + response.body); +} + +void test_request_validation_error() { + // Test that RequestValidationError gets converted to a HTTP 400 + Fixture f( + [](const rt::TaskRequest &) -> rt::TaskResult { + throw rt::RequestValidationError("steps must be positive"); + } + ); + f.add("test-validation"); + const auto response = f.post("/v1/tasks/run", + R"({"model":"test-validation", "text":"hello"})"); + require(response.status == 400, "Expected HTTP 400, got " + std::to_string(response.status) + ": " + response.body); + require(response.body.find("steps must be positive") != std::string::npos, "HTTP response did not contain the expected error: " + response.body); + require(response.body.find("invalid_request_error") != std::string::npos, "HTTP response did not identify invalid_request_error: " + response.body); +} + +void test_insufficient_memory_error() { + // Test that InsufficientMemoryError gets converted to a HTTP 503 + Fixture f( + [](const rt::TaskRequest &) -> rt::TaskResult { + throw srv::InsufficientMemoryError("VRAM oom"); + } + ); + f.add("test-memory"); + const auto response = f.post("/v1/tasks/run", + R"({"model":"test-memory", "text":"hello"})"); + require(response.status == 503, "Expected HTTP 503, got " + std::to_string(response.status) + ": " + response.body); + require(response.body.find("VRAM oom") != std::string::npos, "HTTP response did not contain the expected error: " + response.body); + require(response.body.find("insufficient_memory") != std::string::npos, "HTTP response did not identify insufficient_memory: " + response.body); +} + +void test_server_busy_error() { + // Test that ServerBusyError gets converted to a HTTP 503 + Fixture f( + [](const rt::TaskRequest &) -> rt::TaskResult { + throw srv::ServerBusyError("busy"); + } + ); + f.add("test-busy"); + const auto response = f.post("/v1/tasks/run", + R"({"model":"test-busy", "text":"hello"})"); + require(response.status == 503, "Expected HTTP 503, got " + std::to_string(response.status) + ": " + response.body); + require(response.body.find("busy") != std::string::npos, "HTTP response did not contain the expected error: " + response.body); + require(response.body.find("server_busy") != std::string::npos, "HTTP response did identify server_busy: " + response.body); +} + +// Need a distinct exception type to ServerState::handle_request's +// std::exception catch block - do not want to reuse any existing exception types +// or any future exception types (so if more narrow catch's get added, this test +// will still test the std::exception catch block), but we also need to attach +// information to ensure it's actually the exception we threw. +class TestException final : public std::exception { +public: + explicit TestException(const char * message) : message_(message) { } + const char * what() const noexcept { return message_.c_str(); } + +private: + std::string message_; +}; + +void test_server_exception_cors() { + // Test that other exception types convert to an HTTP 500 when from an allowed origin + Fixture f( + [](const rt::TaskRequest &) -> rt::TaskResult { + throw TestException("internal error"); + } + ); + f.add("test-cors1"); + const auto response = f.post("/v1/tasks/run", + R"({"model":"test-cors1", "text":"hello"})", "http://allowed-origin"); + require(response.status == 500, "Expected HTTP 500, got " + std::to_string(response.status) + ": " + response.body); + require(response.body.find("internal error") != std::string::npos, "HTTP response did not contain the expected error: " + response.body); + require(response.body.find("server_error") != std::string::npos, "HTTP response did not identify server_error: " + response.body); + require(response.headers.at("Access-Control-Allow-Origin") == "http://allowed-origin", "Expected Access-Control-Allow-Origin header"); +} + +void test_server_exception_no_cors() { + // Test that the server throws the exception when not from an allowed origin + Fixture f( + [](const rt::TaskRequest &) -> rt::TaskResult { + throw TestException("rethrow"); + }, + "" + ); + f.add("test-cors2"); + try { + const auto response = f.post("/v1/tasks/run", + R"({"model":"test-cors2", "text":"hello"})"); + } catch (const TestException & ex) { + require(std::string(ex.what()) == "rethrow", "caught exception was not the exception thrown"); + return; + } + // The exception should have been caught, if it didn't, it got swallowed somewhere internally. + throw std::runtime_error("CORS exception did not result in an uncaught exception"); +} + +} // namespace + +int main() { + try { + engine::io::json::enable_serialized_json_parsing(); + test_capacity_error(); + test_insufficient_memory_error(); + test_request_validation_error(); + test_server_busy_error(); + test_server_exception_cors(); + test_server_exception_no_cors(); + + std::cout << "PASS ServerState request exception handling\n"; + return 0; + } catch (const std::exception & ex) { + std::cerr << "FAIL: " << ex.what() << '\n'; + return 1; + } +} From 3688317d8a1043f48e4af48514004504cf9cbeb8 Mon Sep 17 00:00:00 2001 From: lojack5 <1458329+lojack5@users.noreply.github.com> Date: Mon, 12 Oct 2026 16:09:15 -0600 Subject: [PATCH 3/3] Test the exception handling of parallel ParallelServerState::handle_request needed to pull out the fixtures from test_server_parallel_lifecycle.cpp to re-use, then modify to inject the tested exceptions --- CMakeLists.txt | 13 + tests/unittests/server_state_fixture.h | 579 +++++++++++++----- .../test_server_parallel_lifecycle.cpp | 254 +------- ...test_server_parallel_request_exception.cpp | 140 +++++ .../test_server_request_exception.cpp | 6 - 5 files changed, 588 insertions(+), 404 deletions(-) create mode 100644 tests/unittests/test_server_parallel_request_exception.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 6b6727a1c..4b0a9c4a2 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -3109,6 +3109,19 @@ if (ENGINE_BUILD_TESTS OR ENGINE_BUILD_EXTENDED_TESTS OR ENGINE_BUILD_MODEL_TEST add_test(NAME server_request_exceptions_test COMMAND server_request_exceptions_test) set_tests_properties(server_request_exceptions_test PROPERTIES TIMEOUT 60) + get_target_property(server_parallel_exception_sources audiocpp_server SOURCES) + list(FILTER server_parallel_exception_sources EXCLUDE REGEX "app/server/(main|model_installer|runtime)\\.cpp$") + add_executable(server_parallel_request_exceptions_test + tests/unittests/test_server_parallel_request_exception.cpp ${server_parallel_exception_sources}) + target_include_directories(server_parallel_request_exceptions_test PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR}/app/server "${CMAKE_CURRENT_BINARY_DIR}/generated") + target_link_libraries(server_parallel_request_exceptions_test PRIVATE engine_runtime ggml Threads::Threads) + if(WIN32) + target_link_libraries(server_parallel_request_exceptions_test PRIVATE ws2_32) + endif() + add_test(NAME server_parallel_request_exceptions_test COMMAND server_parallel_request_exceptions_test) + set_tests_properties(server_parallel_request_exceptions_test PROPERTIES TIMEOUT 60) + add_executable(session_pool_test tests/unittests/test_session_pool.cpp) target_include_directories(session_pool_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include ${CMAKE_CURRENT_SOURCE_DIR}/app/server) # TaskResult owns JSON values for custom-schema outputs. diff --git a/tests/unittests/server_state_fixture.h b/tests/unittests/server_state_fixture.h index cc1418156..beb9094b0 100644 --- a/tests/unittests/server_state_fixture.h +++ b/tests/unittests/server_state_fixture.h @@ -1,181 +1,464 @@ #pragma once -// Define Mock server, model, and model loaders to use as test fixtures -// when testing ServerState specific code that doesn't require a model -// actually loaded, nor a server started and bound to any addresses +#ifdef SERVER_STATE_FIXTURE_PARALLEL + #include "parallel_runtime.h" + #include "engine/framework/audio/wav_writer.h" -#include "runtime.h" -#include "engine/framework/runtime/registry.h" + #include + #include + #include + #include + #include + #include +#else + #include "runtime.h" + #include "engine/framework/runtime/registry.h" + + #include + #include + #include +#endif #include -#include -#include #include -#include +#include #include #include -#include namespace rt = engine::runtime; namespace srv = minitts::server; - namespace { +void require(bool condition, const std::string & message) { + if (!condition) { throw std::runtime_error(message); } +} + // lambda for test-supplied model run behavior using RunFunction = std::function; -inline constexpr const char * kModelFamily = "test-server-fixture"; - -// Minimal mock Offline Session -class MockSession final : public rt::IOfflineVoiceTaskSession { -public: - MockSession(const rt::TaskSpec & task, std::shared_ptr run) - : task_(task), run_(std::move(run)) {} - - std::string family() const override { return kModelFamily; } - rt::VoiceTaskKind task_kind() const override { return task_.task; } - rt::RunMode run_mode() const override { return task_.mode; } - void prepare(const rt::SessionPreparationRequest &) override {} - rt::TaskResult run(const rt::TaskRequest & request) override { return (*run_)(request); } - -private: - rt::TaskSpec task_; - std::shared_ptr run_; -}; - -// Minimal mock offline model -class MockModel final : public rt::ILoadedVoiceModel { -public: - explicit MockModel(std::shared_ptr run) - : run_(std::move(run)) { - metadata_.family = kModelFamily; - } +} // namespace - const rt::ModelMetadata & metadata() const noexcept override { return metadata_; } - const rt::CapabilitySet & capabilities() const noexcept override { return capabilities_; } - std::unique_ptr create_task_session(const rt::TaskSpec & task, const rt::SessionOptions &) const { - return std::make_unique(task, run_); - } -private: - std::shared_ptr run_; - rt::ModelMetadata metadata_; - rt::CapabilitySet capabilities_; -}; - -// Mock model loader so ServerState can "load" our mock model without any actual weights -class MockLoader final : public rt::IVoiceModelLoader { -public: - explicit MockLoader(std::shared_ptr run) : run_(std::move(run)) {} - - std::string family() const override { return kModelFamily; } - bool can_load(const rt::ModelLoadRequest &) const override { return true; } - rt::ModelInspection inspect(const rt::ModelLoadRequest &) const override { return {}; } - std::unique_ptr load(const rt::ModelLoadRequest &) const override { - return std::make_unique(run_); +// Define Mock server, model, and model loaders to use as test fixtures +// when testing ServerState specific code that doesn't require a model +// actually loaded, nor a server started and bound to any addresses +#ifdef SERVER_STATE_FIXTURE_PARALLEL + using namespace std::chrono_literals; + using engine::io::json::parse; + + namespace { + template bool observe(F f) { + const auto deadline = std::chrono::steady_clock::now() + 2s; + while (!f()) { + if (std::chrono::steady_clock::now() >= deadline) { return false; } + std::this_thread::yield(); + } + return true; } + struct Gate { + std::mutex mutex; + std::condition_variable cv; + bool enabled = false, entered = false, released = false; + void arm() { std::lock_guard lock(mutex); enabled = true; entered = released = false; } + void wait() { + std::unique_lock lock(mutex); + if (!enabled) { return; } + entered = true; + cv.notify_all(); + // A failing test must still drain its futures and release server state. + cv.wait_for(lock, 5s, [&] { return released; }); + } + bool seen() { std::lock_guard lock(mutex); return entered; } + void release() { std::lock_guard lock(mutex); released = true; cv.notify_all(); } + }; + struct Control { + Gate load, run, destroy; + std::atomic loads{0}, destructions{0}, runs{0}; + std::atomic sessions{0}, clones{0}; + std::atomic fail_load{false}, null_session{false}, fail_prepare{false}, fail_run{false}, fail_clone{false}; + std::atomic premature_model_destruction{false}; + }; + class Session final : public rt::IOfflineVoiceTaskSession, + public rt::IBatchedOfflineVoiceTaskSession, + public rt::IStreamingVoiceTaskSession, + public rt::IParallelVoiceTaskSessionFactory { + public: + Session(std::shared_ptr c, rt::TaskSpec task, std::string revision, size_t capacity = 4, std::shared_ptr run = {}) + : control_(std::move(c)), run_(std::move(run)), task_(task), revision_(std::move(revision)), capacity_(capacity) { ++control_->sessions; } + ~Session() override { --control_->sessions; } + std::string family() const override { return "lifecycle_fixture"; } + rt::VoiceTaskKind task_kind() const override { return task_.task; } + rt::RunMode run_mode() const override { return task_.mode; } + void prepare(const rt::SessionPreparationRequest &) override { + if (control_->fail_prepare) { throw std::runtime_error("controlled preparation failure"); } + } + size_t parallel_session_capacity() const noexcept override { return capacity_; } + std::unique_ptr create_parallel_session() const override { + if (++control_->clones == 2 && control_->fail_clone) { + throw std::runtime_error("controlled partial pool failure"); + } + return std::make_unique(control_, task_, revision_, capacity_, run_); + } + rt::TaskResult run(const rt::TaskRequest & request) override { + if (run_ && *run_) { + return (*run_)(request); + } else { + return run_default(request); + } + } -private: - std::shared_ptr run_; -}; -} // namespace + rt::TaskResult run_default(const rt::TaskRequest & request) { + control_->run.wait(); + ++control_->runs; + if (control_->fail_run) { throw std::runtime_error("controlled inference failure"); } + if (task_.task == rt::VoiceTaskKind::Asr && !request.audio_input) { + throw std::runtime_error("replacement ASR requires audio input"); + } + rt::TaskResult result; + result.text_output = rt::Transcript{revision_, "en"}; + result.audio_output = rt::AudioBuffer{16000, 1, {0.1f, -0.1f}}; + return result; + } + std::vector run_batch(const std::vector & requests) override { + std::vector results; + for (const auto & request : requests) { results.push_back(run(request)); } + return results; + } + rt::StreamingPolicy streaming_policy() const override { + rt::StreamingPolicy policy; + policy.input = rt::StreamingInputKind::None; + policy.output = rt::StreamingOutputKind::PullEvents; + return policy; + } + void start_stream(const rt::TaskRequest & request) override { result_ = run(request); emitted_ = false; } + std::optional next_stream_event() override { + if (emitted_) { return {}; } + emitted_ = true; + rt::StreamEvent event; + event.audio_output = result_.audio_output; + event.partial_text = result_.text_output; + event.is_final = true; + return event; + } + void reset() override { emitted_ = false; } + rt::StreamEvent process_audio_chunk(const rt::AudioChunk &) override { return {}; } + rt::TaskResult finalize() override { return result_; } + private: + std::shared_ptr control_; + std::shared_ptr run_; + rt::TaskSpec task_; + std::string revision_; + size_t capacity_; + rt::TaskResult result_; + bool emitted_ = false; + }; + class Model final : public rt::ILoadedVoiceModel { + public: + Model(std::shared_ptr c, std::string revision, std::shared_ptr run = {}) + : control_(std::move(c)), run_(std::move(run)), revision_(std::move(revision)) { + metadata_.family = "lifecycle_fixture"; + } + ~Model() override { + control_->destroy.wait(); + if (control_->sessions != 0) { control_->premature_model_destruction = true; } + ++control_->destructions; + } + const rt::ModelMetadata & metadata() const noexcept override { return metadata_; } + const rt::CapabilitySet & capabilities() const noexcept override { return capabilities_; } + std::unique_ptr create_task_session(const rt::TaskSpec & task, const rt::SessionOptions & options) const override { + if (control_->null_session) { return nullptr; } + const auto cap = options.options.find("fixture_capacity"); + return std::make_unique(control_, task, revision_, + cap == options.options.end() ? 4 : std::stoul(cap->second), + run_); + } + private: + std::shared_ptr control_; + std::shared_ptr run_; + std::string revision_; + rt::ModelMetadata metadata_; + rt::CapabilitySet capabilities_; + }; + class Loader final : public rt::IVoiceModelLoader { + public: + explicit Loader(std::shared_ptr run = {}): run_(std::move(run)) { } -// Test-only friend class of runtime::ServerState, needed -// to modify its model loader to use the mock loader. -// Declared in the minitts::server namespace originally -namespace minitts::server { - -class ServerRuntimeTestAccess { -public: - static void set_registry_factory(ServerState & state, const std::shared_ptr & loader) { - state.registry_factory_ = [loader] { - rt::ModelRegistry registry; - registry.register_loader(loader); - return registry; - }; - } + std::map> controls; + std::string family() const override { return "lifecycle_fixture"; } + bool can_load(const rt::ModelLoadRequest &) const override { return true; } + rt::ModelInspection inspect(const rt::ModelLoadRequest &) const override { return {}; } + std::unique_ptr load(const rt::ModelLoadRequest & request) const override { + const auto c = controls.at(request.model_path.filename().string()); + ++c->loads; + c->load.wait(); + if (c->fail_load) { throw std::runtime_error("controlled load failure"); } + return std::make_unique(c, request.model_path.filename().string(), run_); + } - static void add(ServerState & state, ServerModelConfig config) { - auto model = state.make_model(std::move(config)); - state.model_index_.emplace(model->config.id, state.models_.size()); - state.models_.push_back(std::move(model)); + private: + std::shared_ptr run_; + }; + std::string quote(const std::string & text) { return engine::io::json::stringify_string(text); } + srv::HttpResponse post(srv::ParallelServerState & state, const std::string & path, std::string body, const std::string & origin = "") { + srv::HttpRequest request; + request.method = "POST"; request.path = path; request.body = std::move(body); + request.headers["content-type"] = "application/json"; + if (!origin.empty()) request.headers["origin"] = origin; + return state.handle(request); } -}; -} // namespace minitts::server -using Access = srv::ServerRuntimeTestAccess; - -namespace { -// Test fixture: test a real ServerState object using a lazily loaded MockModel, so -// no weights are loaded, no actual server is launched. -class Fixture { -public: - explicit Fixture(RunFunction run, std::string cors = "*") - : root_(make_temp_root()), - run_(std::make_shared(std::move(run))), - loader_(std::make_shared(run_)) { - srv::ServerConfig config; - config.backend = engine::core::BackendType::Cpu; - config.ui_enabled = false; - config.ui_management = true; // needed to ensure the model loader will use our mock model - config.idle_unload_ms = 0; - config.max_loaded_models = 1; - config.min_free_memory_mb = 0; - config.cors_origins = cors; - state_ = std::make_unique(std::move(config), root_); - Access::set_registry_factory(*state_, loader_); + srv::HttpResponse run(srv::ParallelServerState & state, const std::string & id) { + return post(state, "/v1/tasks/run", "{\"model\":" + quote(id) + ",\"text\":\"old request\"}"); } - Fixture(const Fixture &) = delete; - ~Fixture() { - // Destroy ServerState before cleaning up temp directory - state_.reset(); - std::error_code ec; - std::filesystem::remove_all(root_, ec); + void success(const srv::HttpResponse & response) { + require(response.status == 200, "HTTP " + std::to_string(response.status) + ": " + response.body); } - Fixture & operator= (const Fixture &) = delete; - - void add(const std::string & id, const std::string & mode = "offline") { - // ServerState's load_model needs a fake file in place when looking up model - // contracts for the fake model - std::ofstream(root_ / id) << "controlled fixture"; - // Now create and add the model, already "loaded" - srv::ServerModelConfig model; - model.id = id; - model.path = root_ / id; - model.family = loader_->family(); - model.mode = mode; - model.lazy = true; - Access::add(*state_, std::move(model)); + class Writer final : public srv::HttpStreamWriter { + public: + std::string output; + bool disconnected = false; + void write(std::string_view s) override { + if (disconnected) { throw std::runtime_error("client disconnected"); } + output.append(s); + } + }; } - // Test convenience methods - srv::HttpResponse handle(const srv::HttpRequest & request) { - return state_->handle(request); + namespace minitts::server { + class ServerRuntimeTestAccess { + public: + static void registry(ParallelServerState & state, const std::shared_ptr & loader) { + state.registry_factory_ = [loader] { + rt::ModelRegistry registry; registry.register_loader(loader); return registry; + }; + state.config_.ui_management = true; + } + static void add(ParallelServerState & state, ServerModelConfig config) { + auto model = state.make_model(std::move(config)); + state.model_index_.emplace(model->registered_id, state.models_.size()); + state.models_.push_back(std::move(model)); + } + static auto & model(ParallelServerState & state, const std::string & id) { + return *state.models_.at(state.model_index_.at(id)); + } + static ModelSlots::State slots(ParallelServerState & state, const std::string & id) { return model(state, id).busy.state(); } + static bool resident(ParallelServerState & state, const std::string & id) { return model(state, id).resident(); } + static bool loaded(ParallelServerState & state, const std::string & id) { return model(state, id).loaded.load(); } + static auto block_metadata(ParallelServerState & state, const std::string & id) { + return std::unique_lock(model(state, id).metadata_mutex); + } + static auto block_bulk_publication(ParallelServerState & state) { + return std::unique_lock(state.bulk_admission_mutex_); + } + }; } + using Access = srv::ServerRuntimeTestAccess; - srv::HttpResponse post(const std::string & path, std::string body, const std::string origin = "http://server-state-test") { - srv::HttpRequest request; - request.method = "POST"; - request.path = path; - request.body = std::move(body); - request.headers["content-type"] = "application/json"; - request.headers["origin"] = origin; - return handle(request); - } + namespace { + struct Fixture { + std::filesystem::path root; + std::shared_ptr run_; + std::shared_ptr loader; + std::unique_ptr state; + int count; + Fixture(int slots, int limit = 0, RunFunction run = nullptr, const std::string cors = "*") + : run_(std::make_shared(std::move(run))), + loader(std::make_shared(std::move(run_))), + count(slots) { + root = std::filesystem::temp_directory_path() / + ("audiocpp-lifecycle-" + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count())); + std::filesystem::create_directories(root); + srv::ServerConfig config; config.backend = engine::core::BackendType::Cpu; + config.ui_enabled = false; config.max_loaded_models = limit; config.busy_timeout_ms = 0; + config.cors_origins = cors; + state = std::make_unique(config, root); + Access::registry(*state, loader); + } + ~Fixture() { + for (const auto & [name, c] : loader->controls) { + c->load.release(); c->run.release(); c->destroy.release(); + } + state.reset(); + std::error_code ec; std::filesystem::remove_all(root, ec); + } + std::shared_ptr asset(const std::string & name) { + std::ofstream(root / name) << "controlled fixture"; + auto c = std::make_shared(); loader->controls.emplace(name, c); return c; + } + std::shared_ptr add(const std::string & id, const std::string & mode = "offline") { + auto c = asset(id); + srv::ServerModelConfig model; + model.id = id; model.path = root / id; model.family = loader->family(); + model.slots = count; model.lazy = true; model.mode = mode; + Access::add(*state, model); return c; + } + auto queued(const std::string & id, int n, bool management = false) { + return observe([&] { const auto s = Access::slots(*state, id); + return (management ? s.waiting_management : s.waiting_requests) == n; }); + } + srv::HttpResponse replace(const std::string & id, const std::string & asset, const std::string & task = "tts") { + return post(*state, "/v1/models/load", "{\"id\":" + quote(id) + ",\"path\":" + + quote((root / asset).generic_string()) + ",\"family\":\"lifecycle_fixture\",\"task\":" + + quote(task) + ",\"slots\":" + std::to_string(count) + "}"); + } + }; -private: - static std::filesystem::path make_temp_root() { - const auto ticks = std::chrono::steady_clock::now().time_since_epoch().count(); - auto root = std::filesystem::temp_directory_path() / - ("audio-cpp-server-state-test-" + std::to_string(ticks)); - std::filesystem::create_directories(root); - return root; - } + } // namespace +#else // Non-parallel, ServerState fixture + namespace { - std::filesystem::path root_; - std::shared_ptr run_; - std::shared_ptr loader_; - std::unique_ptr state_; -}; + inline constexpr const char * kModelFamily = "test-server-fixture"; -} // namespace + // Minimal mock Offline Session + class MockSession final : public rt::IOfflineVoiceTaskSession { + public: + MockSession(const rt::TaskSpec & task, std::shared_ptr run) + : task_(task), run_(std::move(run)) {} + + std::string family() const override { return kModelFamily; } + rt::VoiceTaskKind task_kind() const override { return task_.task; } + rt::RunMode run_mode() const override { return task_.mode; } + void prepare(const rt::SessionPreparationRequest &) override {} + rt::TaskResult run(const rt::TaskRequest & request) override { return (*run_)(request); } + + private: + rt::TaskSpec task_; + std::shared_ptr run_; + }; + + // Minimal mock offline model + class MockModel final : public rt::ILoadedVoiceModel { + public: + explicit MockModel(std::shared_ptr run) + : run_(std::move(run)) { + metadata_.family = kModelFamily; + } + + const rt::ModelMetadata & metadata() const noexcept override { return metadata_; } + const rt::CapabilitySet & capabilities() const noexcept override { return capabilities_; } + std::unique_ptr create_task_session(const rt::TaskSpec & task, const rt::SessionOptions &) const { + return std::make_unique(task, run_); + } + + private: + std::shared_ptr run_; + rt::ModelMetadata metadata_; + rt::CapabilitySet capabilities_; + }; + + // Mock model loader so ServerState can "load" our mock model without any actual weights + class MockLoader final : public rt::IVoiceModelLoader { + public: + explicit MockLoader(std::shared_ptr run) : run_(std::move(run)) {} + + std::string family() const override { return kModelFamily; } + bool can_load(const rt::ModelLoadRequest &) const override { return true; } + rt::ModelInspection inspect(const rt::ModelLoadRequest &) const override { return {}; } + std::unique_ptr load(const rt::ModelLoadRequest &) const override { + return std::make_unique(run_); + } + + private: + std::shared_ptr run_; + }; + } // namespace + + // Test-only friend class of runtime::ServerState, needed + // to modify its model loader to use the mock loader. + // Declared in the minitts::server namespace originally + namespace minitts::server { + + class ServerRuntimeTestAccess { + public: + static void set_registry_factory(ServerState & state, const std::shared_ptr & loader) { + state.registry_factory_ = [loader] { + rt::ModelRegistry registry; + registry.register_loader(loader); + return registry; + }; + } + + static void add(ServerState & state, ServerModelConfig config) { + auto model = state.make_model(std::move(config)); + state.model_index_.emplace(model->config.id, state.models_.size()); + state.models_.push_back(std::move(model)); + } + }; + } // namespace minitts::server + using Access = srv::ServerRuntimeTestAccess; + + namespace { + // Test fixture: test a real ServerState object using a lazily loaded MockModel, so + // no weights are loaded, no actual server is launched. + class Fixture { + public: + explicit Fixture(RunFunction run, std::string cors = "*") + : root_(make_temp_root()), + run_(std::make_shared(std::move(run))), + loader_(std::make_shared(run_)) { + srv::ServerConfig config; + config.backend = engine::core::BackendType::Cpu; + config.ui_enabled = false; + config.ui_management = true; // needed to ensure the model loader will use our mock model + config.idle_unload_ms = 0; + config.max_loaded_models = 1; + config.min_free_memory_mb = 0; + config.cors_origins = cors; + state_ = std::make_unique(std::move(config), root_); + Access::set_registry_factory(*state_, loader_); + } + Fixture(const Fixture &) = delete; + ~Fixture() { + // Destroy ServerState before cleaning up temp directory + state_.reset(); + std::error_code ec; + std::filesystem::remove_all(root_, ec); + } + Fixture & operator= (const Fixture &) = delete; + + void add(const std::string & id, const std::string & mode = "offline") { + // ServerState's load_model needs a fake file in place when looking up model + // contracts for the fake model + std::ofstream(root_ / id) << "controlled fixture"; + // Now create and add the model, already "loaded" + srv::ServerModelConfig model; + model.id = id; + model.path = root_ / id; + model.family = loader_->family(); + model.mode = mode; + model.lazy = true; + Access::add(*state_, std::move(model)); + } + + // Test convenience methods + srv::HttpResponse handle(const srv::HttpRequest & request) { + return state_->handle(request); + } + + srv::HttpResponse post(const std::string & path, std::string body, const std::string origin = "http://server-state-test") { + srv::HttpRequest request; + request.method = "POST"; + request.path = path; + request.body = std::move(body); + request.headers["content-type"] = "application/json"; + request.headers["origin"] = origin; + return handle(request); + } + + private: + static std::filesystem::path make_temp_root() { + const auto ticks = std::chrono::steady_clock::now().time_since_epoch().count(); + auto root = std::filesystem::temp_directory_path() / + ("audio-cpp-server-state-test-" + std::to_string(ticks)); + std::filesystem::create_directories(root); + return root; + } + + std::filesystem::path root_; + std::shared_ptr run_; + std::shared_ptr loader_; + std::unique_ptr state_; + }; + + } // namespace +#endif diff --git a/tests/unittests/test_server_parallel_lifecycle.cpp b/tests/unittests/test_server_parallel_lifecycle.cpp index e540ddbd2..222229a2d 100644 --- a/tests/unittests/test_server_parallel_lifecycle.cpp +++ b/tests/unittests/test_server_parallel_lifecycle.cpp @@ -1,25 +1,10 @@ // Run the actual server handlers with tiny controlled loaders. Barriers force // lifecycle overlaps without GPU timing, sleeps in production, or large models. -#include "parallel_runtime.h" -#include "engine/framework/audio/wav_writer.h" -#include -#include -#include -#include -#include -#include -#include -#include - -using namespace std::chrono_literals; -namespace rt = engine::runtime; -namespace srv = minitts::server; -using engine::io::json::parse; +#define SERVER_STATE_FIXTURE_PARALLEL +#include "server_state_fixture.h" namespace { -void require(bool value, const std::string & message) { - if (!value) { throw std::runtime_error(message); } -} + void concurrent_json_parsing() { std::promise start; auto ready = start.get_future().share(); @@ -47,237 +32,6 @@ void concurrent_json_parsing() { start.set_value(); for (auto & worker : workers) { worker.get(); } } -template bool observe(F f) { - const auto deadline = std::chrono::steady_clock::now() + 2s; - while (!f()) { - if (std::chrono::steady_clock::now() >= deadline) { return false; } - std::this_thread::yield(); - } - return true; -} -struct Gate { - std::mutex mutex; - std::condition_variable cv; - bool enabled = false, entered = false, released = false; - void arm() { std::lock_guard lock(mutex); enabled = true; entered = released = false; } - void wait() { - std::unique_lock lock(mutex); - if (!enabled) { return; } - entered = true; - cv.notify_all(); - // A failing test must still drain its futures and release server state. - cv.wait_for(lock, 5s, [&] { return released; }); - } - bool seen() { std::lock_guard lock(mutex); return entered; } - void release() { std::lock_guard lock(mutex); released = true; cv.notify_all(); } -}; -struct Control { - Gate load, run, destroy; - std::atomic loads{0}, destructions{0}, runs{0}; - std::atomic sessions{0}, clones{0}; - std::atomic fail_load{false}, null_session{false}, fail_prepare{false}, fail_run{false}, fail_clone{false}; - std::atomic premature_model_destruction{false}; -}; -class Session final : public rt::IOfflineVoiceTaskSession, - public rt::IBatchedOfflineVoiceTaskSession, - public rt::IStreamingVoiceTaskSession, - public rt::IParallelVoiceTaskSessionFactory { -public: - Session(std::shared_ptr c, rt::TaskSpec task, std::string revision, size_t capacity = 4) - : control_(std::move(c)), task_(task), revision_(std::move(revision)), capacity_(capacity) { ++control_->sessions; } - ~Session() override { --control_->sessions; } - std::string family() const override { return "lifecycle_fixture"; } - rt::VoiceTaskKind task_kind() const override { return task_.task; } - rt::RunMode run_mode() const override { return task_.mode; } - void prepare(const rt::SessionPreparationRequest &) override { - if (control_->fail_prepare) { throw std::runtime_error("controlled preparation failure"); } - } - size_t parallel_session_capacity() const noexcept override { return capacity_; } - std::unique_ptr create_parallel_session() const override { - if (++control_->clones == 2 && control_->fail_clone) { - throw std::runtime_error("controlled partial pool failure"); - } - return std::make_unique(control_, task_, revision_, capacity_); - } - rt::TaskResult run(const rt::TaskRequest & request) override { - control_->run.wait(); - ++control_->runs; - if (control_->fail_run) { throw std::runtime_error("controlled inference failure"); } - if (task_.task == rt::VoiceTaskKind::Asr && !request.audio_input) { - throw std::runtime_error("replacement ASR requires audio input"); - } - rt::TaskResult result; - result.text_output = rt::Transcript{revision_, "en"}; - result.audio_output = rt::AudioBuffer{16000, 1, {0.1f, -0.1f}}; - return result; - } - std::vector run_batch(const std::vector & requests) override { - std::vector results; - for (const auto & request : requests) { results.push_back(run(request)); } - return results; - } - rt::StreamingPolicy streaming_policy() const override { - rt::StreamingPolicy policy; - policy.input = rt::StreamingInputKind::None; - policy.output = rt::StreamingOutputKind::PullEvents; - return policy; - } - void start_stream(const rt::TaskRequest & request) override { result_ = run(request); emitted_ = false; } - std::optional next_stream_event() override { - if (emitted_) { return {}; } - emitted_ = true; - rt::StreamEvent event; - event.audio_output = result_.audio_output; - event.partial_text = result_.text_output; - event.is_final = true; - return event; - } - void reset() override { emitted_ = false; } - rt::StreamEvent process_audio_chunk(const rt::AudioChunk &) override { return {}; } - rt::TaskResult finalize() override { return result_; } -private: - std::shared_ptr control_; - rt::TaskSpec task_; - std::string revision_; - size_t capacity_; - rt::TaskResult result_; - bool emitted_ = false; -}; -class Model final : public rt::ILoadedVoiceModel { -public: - Model(std::shared_ptr c, std::string revision) : control_(std::move(c)), revision_(std::move(revision)) { - metadata_.family = "lifecycle_fixture"; - } - ~Model() override { - control_->destroy.wait(); - if (control_->sessions != 0) { control_->premature_model_destruction = true; } - ++control_->destructions; - } - const rt::ModelMetadata & metadata() const noexcept override { return metadata_; } - const rt::CapabilitySet & capabilities() const noexcept override { return capabilities_; } - std::unique_ptr create_task_session(const rt::TaskSpec & task, const rt::SessionOptions & options) const override { - if (control_->null_session) { return nullptr; } - const auto cap = options.options.find("fixture_capacity"); - return std::make_unique(control_, task, revision_, - cap == options.options.end() ? 4 : std::stoul(cap->second)); - } -private: - std::shared_ptr control_; - std::string revision_; - rt::ModelMetadata metadata_; - rt::CapabilitySet capabilities_; -}; -class Loader final : public rt::IVoiceModelLoader { -public: - std::map> controls; - std::string family() const override { return "lifecycle_fixture"; } - bool can_load(const rt::ModelLoadRequest &) const override { return true; } - rt::ModelInspection inspect(const rt::ModelLoadRequest &) const override { return {}; } - std::unique_ptr load(const rt::ModelLoadRequest & request) const override { - const auto c = controls.at(request.model_path.filename().string()); - ++c->loads; - c->load.wait(); - if (c->fail_load) { throw std::runtime_error("controlled load failure"); } - return std::make_unique(c, request.model_path.filename().string()); - } -}; -std::string quote(const std::string & text) { return engine::io::json::stringify_string(text); } -srv::HttpResponse post(srv::ParallelServerState & state, const std::string & path, std::string body) { - srv::HttpRequest request; - request.method = "POST"; request.path = path; request.body = std::move(body); - request.headers["content-type"] = "application/json"; - return state.handle(request); -} -srv::HttpResponse run(srv::ParallelServerState & state, const std::string & id) { - return post(state, "/v1/tasks/run", "{\"model\":" + quote(id) + ",\"text\":\"old request\"}"); -} -void success(const srv::HttpResponse & response) { - require(response.status == 200, "HTTP " + std::to_string(response.status) + ": " + response.body); -} -class Writer final : public srv::HttpStreamWriter { -public: - std::string output; - bool disconnected = false; - void write(std::string_view s) override { - if (disconnected) { throw std::runtime_error("client disconnected"); } - output.append(s); - } -}; -} - -namespace minitts::server { -class ServerRuntimeTestAccess { -public: - static void registry(ParallelServerState & state, const std::shared_ptr & loader) { - state.registry_factory_ = [loader] { - rt::ModelRegistry registry; registry.register_loader(loader); return registry; - }; - state.config_.ui_management = true; - } - static void add(ParallelServerState & state, ServerModelConfig config) { - auto model = state.make_model(std::move(config)); - state.model_index_.emplace(model->registered_id, state.models_.size()); - state.models_.push_back(std::move(model)); - } - static auto & model(ParallelServerState & state, const std::string & id) { - return *state.models_.at(state.model_index_.at(id)); - } - static ModelSlots::State slots(ParallelServerState & state, const std::string & id) { return model(state, id).busy.state(); } - static bool resident(ParallelServerState & state, const std::string & id) { return model(state, id).resident(); } - static bool loaded(ParallelServerState & state, const std::string & id) { return model(state, id).loaded.load(); } - static auto block_metadata(ParallelServerState & state, const std::string & id) { - return std::unique_lock(model(state, id).metadata_mutex); - } - static auto block_bulk_publication(ParallelServerState & state) { - return std::unique_lock(state.bulk_admission_mutex_); - } -}; -} -using Access = srv::ServerRuntimeTestAccess; - -namespace { -struct Fixture { - std::filesystem::path root; - std::shared_ptr loader = std::make_shared(); - std::unique_ptr state; - int count; - Fixture(int slots, int limit = 0) : count(slots) { - root = std::filesystem::temp_directory_path() / - ("audiocpp-lifecycle-" + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count())); - std::filesystem::create_directories(root); - srv::ServerConfig config; config.backend = engine::core::BackendType::Cpu; - config.ui_enabled = false; config.max_loaded_models = limit; config.busy_timeout_ms = 0; - state = std::make_unique(config, root); - Access::registry(*state, loader); - } - ~Fixture() { - for (const auto & [name, c] : loader->controls) { - c->load.release(); c->run.release(); c->destroy.release(); - } - state.reset(); - std::error_code ec; std::filesystem::remove_all(root, ec); - } - std::shared_ptr asset(const std::string & name) { - std::ofstream(root / name) << "controlled fixture"; - auto c = std::make_shared(); loader->controls.emplace(name, c); return c; - } - std::shared_ptr add(const std::string & id, const std::string & mode = "offline") { - auto c = asset(id); - srv::ServerModelConfig model; - model.id = id; model.path = root / id; model.family = loader->family(); - model.slots = count; model.lazy = true; model.mode = mode; - Access::add(*state, model); return c; - } - auto queued(const std::string & id, int n, bool management = false) { - return observe([&] { const auto s = Access::slots(*state, id); - return (management ? s.waiting_management : s.waiting_requests) == n; }); - } - srv::HttpResponse replace(const std::string & id, const std::string & asset, const std::string & task = "tts") { - return post(*state, "/v1/models/load", "{\"id\":" + quote(id) + ",\"path\":" + - quote((root / asset).generic_string()) + ",\"family\":\"lifecycle_fixture\",\"task\":" + - quote(task) + ",\"slots\":" + std::to_string(count) + "}"); - } -}; void older_work_before_management(int count, bool reconfigure) { Fixture f(count); auto c = f.add("shared"); f.asset("replacement"); @@ -660,7 +414,7 @@ void bulk_publication_excludes_only_other_bulk_calls(int count) { publication.unlock(); success(first.get()); success(second.get()); success(independent.get()); require(excluded && progressed, "bulk publication interleaved barriers or blocked normal inference"); } -} +} // namespace void unsupported_capacity_preserves_registration() { for (int count : {1, 2, 4}) { diff --git a/tests/unittests/test_server_parallel_request_exception.cpp b/tests/unittests/test_server_parallel_request_exception.cpp new file mode 100644 index 000000000..887c3d75f --- /dev/null +++ b/tests/unittests/test_server_parallel_request_exception.cpp @@ -0,0 +1,140 @@ +#define SERVER_STATE_FIXTURE_PARALLEL +#include "server_state_fixture.h" +#include "engine/framework/runtime/errors.h" + +#include +#include +#include + + +namespace { + +void test_capacity_error() { + // Test that CapacityError gets converted to a HTTP 400 + Fixture f(1, 0, + [](const rt::TaskRequest &) -> rt::TaskResult { + throw rt::CapacityError("not enough capacity"); + } + ); + f.add("test-capacity"); + const auto response = post(*f.state, "/v1/tasks/run", + R"({"model":"test-capacity", "text":"hello"})"); + require(response.status == 400, "Expected HTTP 400, got " + std::to_string(response.status) + ": " + response.body); + require(response.body.find("not enough capacity") != std::string::npos, "HTTP response did not contain the expected error: " + response.body); + require(response.body.find("invalid_request_error") != std::string::npos, "HTTP response did not identify invalid_request_error: " + response.body); +} + +void test_request_validation_error() { + // Test that RequestValidationError gets converted to a HTTP 400 + Fixture f(1, 0, + [](const rt::TaskRequest &) -> rt::TaskResult { + throw rt::RequestValidationError("steps must be positive"); + } + ); + f.add("test-validation"); + const auto response = post(*f.state, "/v1/tasks/run", + R"({"model":"test-validation", "text":"hello"})"); + require(response.status == 400, "Expected HTTP 400, got " + std::to_string(response.status) + ": " + response.body); + require(response.body.find("steps must be positive") != std::string::npos, "HTTP response did not contain the expected error: " + response.body); + require(response.body.find("invalid_request_error") != std::string::npos, "HTTP response did not identify invalid_request_error: " + response.body); +} + +void test_insufficient_memory_error() { + // Test that InsufficientMemoryError gets converted to a HTTP 503 + Fixture f(1, 0, + [](const rt::TaskRequest &) -> rt::TaskResult { + throw srv::InsufficientMemoryError("VRAM oom"); + } + ); + f.add("test-memory"); + const auto response = post(*f.state, "/v1/tasks/run", + R"({"model":"test-memory", "text":"hello"})"); + require(response.status == 503, "Expected HTTP 503, got " + std::to_string(response.status) + ": " + response.body); + require(response.body.find("VRAM oom") != std::string::npos, "HTTP response did not contain the expected error: " + response.body); + require(response.body.find("insufficient_memory") != std::string::npos, "HTTP response did not identify insufficient_memory: " + response.body); +} + +void test_server_busy_error() { + // Test that ServerBusyError gets converted to a HTTP 503 + Fixture f(1, 0, + [](const rt::TaskRequest &) -> rt::TaskResult { + throw srv::ServerBusyError("busy"); + } + ); + f.add("test-busy"); + const auto response = post(*f.state, "/v1/tasks/run", + R"({"model":"test-busy", "text":"hello"})"); + require(response.status == 503, "Expected HTTP 503, got " + std::to_string(response.status) + ": " + response.body); + require(response.body.find("busy") != std::string::npos, "HTTP response did not contain the expected error: " + response.body); + require(response.body.find("server_busy") != std::string::npos, "HTTP response did identify server_busy: " + response.body); +} + +// Need a distinct exception type to ServerState::handle_request's +// std::exception catch block - do not want to reuse any existing exception types +// or any future exception types (so if more narrow catch's get added, this test +// will still test the std::exception catch block), but we also need to attach +// information to ensure it's actually the exception we threw. +class TestException final : public std::exception { +public: + explicit TestException(const char * message) : message_(message) { } + const char * what() const noexcept { return message_.c_str(); } + +private: + std::string message_; +}; + +void test_server_exception_cors() { + // Test that other exception types convert to an HTTP 500 when from an allowed origin + Fixture f(1, 0, + [](const rt::TaskRequest &) -> rt::TaskResult { + throw TestException("internal error"); + } + ); + f.add("test-cors1"); + const auto response = post(*f.state, "/v1/tasks/run", + R"({"model":"test-cors1", "text":"hello"})", "http://allowed-origin"); + require(response.status == 500, "Expected HTTP 500, got " + std::to_string(response.status) + ": " + response.body); + require(response.body.find("internal error") != std::string::npos, "HTTP response did not contain the expected error: " + response.body); + require(response.body.find("server_error") != std::string::npos, "HTTP response did not identify server_error: " + response.body); + require(response.headers.at("Access-Control-Allow-Origin") == "http://allowed-origin", "Expected Access-Control-Allow-Origin header"); +} + +void test_server_exception_no_cors() { + // Test that the server throws the exception when not from an allowed origin + Fixture f(1, 0, + [](const rt::TaskRequest &) -> rt::TaskResult { + throw TestException("rethrow"); + }, + "" + ); + f.add("test-cors2"); + try { + const auto response = post(*f.state, "/v1/tasks/run", + R"({"model":"test-cors2", "text":"hello"})"); + } catch (const TestException & ex) { + require(std::string(ex.what()) == "rethrow", "caught exception was not the exception thrown"); + return; + } + // The exception should have been caught, if it didn't, it got swallowed somewhere internally. + throw std::runtime_error("CORS exception did not result in an uncaught exception"); +} + +} // namespace + +int main() { + try { + engine::io::json::enable_serialized_json_parsing(); + test_capacity_error(); + test_insufficient_memory_error(); + test_request_validation_error(); + test_server_busy_error(); + test_server_exception_cors(); + test_server_exception_no_cors(); + + std::cout << "PASS ServerState request exception handling\n"; + return 0; + } catch (const std::exception & ex) { + std::cerr << "FAIL: " << ex.what() << '\n'; + return 1; + } +} diff --git a/tests/unittests/test_server_request_exception.cpp b/tests/unittests/test_server_request_exception.cpp index ddf1d699c..cdc53ecd7 100644 --- a/tests/unittests/test_server_request_exception.cpp +++ b/tests/unittests/test_server_request_exception.cpp @@ -8,12 +8,6 @@ namespace { -void require(bool condition, const std::string & message) { - if (!condition) - throw std::runtime_error(message); -} - - void test_capacity_error() { // Test that CapacityError gets converted to a HTTP 400 Fixture f(