product-vamp/tests/test_score_loader.cpp

#include <boost/test/unit_test.hpp>

#include <algorithm>
#include <atomic>
#include <chrono>
#include <future>
#include <memory>
#include <stdexcept>
#include <thread>
#include <vector>

#include <boost/asio/io_context.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/redirect_error.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/thread_pool.hpp>

#include <vamp_zero/errors.hpp>
#include <vamp_zero/score_loader.hpp>


namespace {

auto performance(std::string id) -> vamp_zero::score_reference {
  return {.id = std::move(id)};
}

auto drain(boost::asio::io_context &io) -> void {
  io.restart();
  io.poll();
}

struct controlled_scores {
  explicit controlled_scores(boost::asio::io_context &io)
      : io{io} {}
  auto load(vamp_zero::score_reference reference) -> boost::asio::awaitable<std::expected<vamp_zero::score_document, modulus::error>> {
    requests.push_back(reference);
    auto timer = std::make_shared<boost::asio::steady_timer>(io, std::chrono::steady_clock::time_point::max());
    waits.push_back(timer);
    maximum_active = std::max(maximum_active, ++active);
    auto error = boost::system::error_code{};
    co_await timer->async_wait(boost::asio::redirect_error(boost::asio::use_awaitable, error));
    --active;
    if (throws) {
      throw std::runtime_error{"fixture failure"};
    }
    if (fails) {
      co_return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::NOT_FOUND, "missing fixture score"));
    }
    co_return vamp_zero::score_document{.id = reference.id, .source_json = "fixture payload"};
  }
  auto complete(std::size_t index) -> void {
    waits.at(index)->cancel();
  }
  boost::asio::io_context &io;
  std::vector<vamp_zero::score_reference> requests{};
  std::vector<std::shared_ptr<boost::asio::steady_timer>> waits{};
  std::size_t active{};
  std::size_t maximum_active{};
  std::bool_t fails{};
  std::bool_t throws{};
};

struct loader_fixture {
  boost::asio::io_context io{};
  std::shared_ptr<controlled_scores> scores{std::make_shared<controlled_scores>(io)};
  std::shared_ptr<vamp_zero::catalog const> catalog{std::make_shared<vamp_zero::catalog const>()};
  std::shared_ptr<std::size_t> completions{std::make_shared<std::size_t>()};
  vamp_zero::score_loader loader{io.get_executor(),
                                 [scores = scores](vamp_zero::score_reference reference) { return scores->load(std::move(reference)); },
                                 [completions = completions] { ++*completions; }};
};

}

BOOST_AUTO_TEST_CASE(score_loader_starts_asynchronously_from_executor_handler) {
  auto fixture = loader_fixture{};
  boost::asio::post(fixture.io, [&] {
    fixture.loader.select(fixture.catalog, performance("a"));
    BOOST_CHECK(fixture.scores->requests.empty());
    BOOST_CHECK(fixture.loader.snapshot().loading);
  });
  drain(fixture.io);
  BOOST_REQUIRE_EQUAL(fixture.scores->requests.size(), 1U);
  BOOST_CHECK(!fixture.loader.snapshot().document);
  fixture.scores->complete(0);
  drain(fixture.io);
  auto loaded = fixture.loader.snapshot();
  BOOST_REQUIRE(loaded.document);
  BOOST_CHECK_EQUAL(loaded.document->id, "a");
  BOOST_CHECK(!loaded.loading);
  BOOST_CHECK(!loaded.error);
  BOOST_CHECK_EQUAL(*fixture.completions, 1U);
}

BOOST_AUTO_TEST_CASE(score_loader_worker_keeps_control_executor_responsive_and_returns_completion) {
  auto io = boost::asio::io_context{};
  auto worker = boost::asio::thread_pool{1};
  auto entered = std::promise<void>{};
  auto release = std::promise<void>{};
  auto resume = release.get_future().share();
  auto completed = false;
  auto control_thread = std::this_thread::get_id();
  auto loader = vamp_zero::score_loader{io.get_executor(),
                                        [&entered, resume](vamp_zero::score_reference reference) -> boost::asio::awaitable<std::expected<vamp_zero::score_document, modulus::error>> {
                                          entered.set_value();
                                          resume.wait();
                                          co_return vamp_zero::score_document{.id = reference.id};
                                        },
                                        [&] { completed = true; BOOST_CHECK(std::this_thread::get_id() == control_thread); }, worker.get_executor()};
  loader.select(std::make_shared<vamp_zero::catalog>(), performance("a"));
  auto started = entered.get_future().wait_for(std::chrono::seconds{2}) == std::future_status::ready;
  auto serviced = false;
  boost::asio::post(io, [&] { serviced = true; });
  io.poll();
  BOOST_CHECK(serviced);
  BOOST_CHECK(!completed);
  release.set_value();
  worker.join();
  io.restart();
  io.run();
  BOOST_REQUIRE(started);
  BOOST_CHECK(completed);
  BOOST_REQUIRE(loader.snapshot().document);
  BOOST_CHECK_EQUAL(loader.snapshot().document->id, "a");
}

BOOST_AUTO_TEST_CASE(score_loader_coalesces_pending_and_discards_stale_results) {
  auto fixture = loader_fixture{};
  fixture.loader.select(fixture.catalog, performance("a"));
  drain(fixture.io);
  fixture.loader.select(fixture.catalog, performance("b"));
  fixture.loader.select(fixture.catalog, performance("c"));
  drain(fixture.io);
  BOOST_REQUIRE_EQUAL(fixture.scores->requests.size(), 1U);
  fixture.scores->complete(0);
  drain(fixture.io);
  BOOST_REQUIRE_EQUAL(fixture.scores->requests.size(), 2U);
  BOOST_CHECK_EQUAL(fixture.scores->requests.back().id, "c");
  BOOST_CHECK(!fixture.loader.snapshot().document);
  BOOST_CHECK(fixture.loader.snapshot().loading);
  BOOST_CHECK_EQUAL(*fixture.completions, 0U);
  fixture.scores->complete(1);
  drain(fixture.io);
  BOOST_REQUIRE(fixture.loader.snapshot().document);
  BOOST_CHECK_EQUAL(fixture.loader.snapshot().document->id, "c");
  BOOST_CHECK_EQUAL(fixture.scores->maximum_active, 1U);
}

BOOST_AUTO_TEST_CASE(score_loader_coalesces_selections_before_executor_starts) {
  auto fixture = loader_fixture{};
  fixture.loader.select(fixture.catalog, performance("a"));
  fixture.loader.select(fixture.catalog, performance("b"));
  fixture.loader.select(fixture.catalog, performance("c"));
  drain(fixture.io);
  BOOST_REQUIRE_EQUAL(fixture.scores->requests.size(), 1U);
  BOOST_CHECK_EQUAL(fixture.scores->requests.front().id, "c");
  fixture.scores->complete(0);
  drain(fixture.io);
  BOOST_REQUIRE(fixture.loader.snapshot().document);
  BOOST_CHECK_EQUAL(fixture.loader.snapshot().document->id, "c");
}

BOOST_AUTO_TEST_CASE(score_loader_stale_error_does_not_replace_pending_selection) {
  auto fixture = loader_fixture{};
  fixture.loader.select(fixture.catalog, performance("missing"));
  drain(fixture.io);
  fixture.loader.select(fixture.catalog, performance("available"));
  fixture.scores->fails = true;
  fixture.scores->complete(0);
  drain(fixture.io);
  BOOST_CHECK(!fixture.loader.snapshot().error);
  BOOST_CHECK(fixture.loader.snapshot().loading);
  BOOST_CHECK_EQUAL(*fixture.completions, 0U);
  fixture.scores->fails = false;
  fixture.scores->complete(1);
  drain(fixture.io);
  BOOST_REQUIRE(fixture.loader.snapshot().document);
  BOOST_CHECK_EQUAL(fixture.loader.snapshot().document->id, "available");
}

BOOST_AUTO_TEST_CASE(score_loader_catalog_identity_invalidates_same_score_selection) {
  auto fixture = loader_fixture{};
  fixture.loader.select(fixture.catalog, performance("same-id"));
  drain(fixture.io);
  fixture.catalog = std::make_shared<vamp_zero::catalog const>();
  fixture.loader.select(fixture.catalog, performance("same-id"));
  fixture.scores->complete(0);
  drain(fixture.io);
  BOOST_REQUIRE_EQUAL(fixture.scores->requests.size(), 2U);
  BOOST_CHECK(!fixture.loader.snapshot().document);
  fixture.scores->complete(1);
  drain(fixture.io);
  BOOST_REQUIRE(fixture.loader.snapshot().document);
}

BOOST_AUTO_TEST_CASE(score_loader_reuses_active_score_and_releases_on_menu_exit) {
  auto fixture = loader_fixture{};
  fixture.loader.select(fixture.catalog, performance("a"));
  drain(fixture.io);
  fixture.scores->complete(0);
  drain(fixture.io);
  auto document = std::weak_ptr{fixture.loader.snapshot().document};
  fixture.loader.select(fixture.catalog, performance("a"));
  drain(fixture.io);
  BOOST_CHECK_EQUAL(fixture.scores->requests.size(), 1U);
  BOOST_CHECK(document.lock() == fixture.loader.snapshot().document);
  fixture.loader.select(fixture.catalog, std::nullopt);
  BOOST_CHECK(document.expired());
  BOOST_CHECK(!fixture.loader.snapshot().loading);
  BOOST_CHECK(!fixture.loader.snapshot().error);
  fixture.loader.select(fixture.catalog, performance("a"));
  drain(fixture.io);
  BOOST_REQUIRE_EQUAL(fixture.scores->requests.size(), 2U);
  fixture.scores->complete(1);
  drain(fixture.io);
}

BOOST_AUTO_TEST_CASE(score_loader_exit_and_reentry_rejects_previous_generation) {
  auto fixture = loader_fixture{};
  fixture.loader.select(fixture.catalog, performance("a"));
  drain(fixture.io);
  fixture.loader.select(fixture.catalog, std::nullopt);
  fixture.loader.select(fixture.catalog, performance("a"));
  fixture.scores->complete(0);
  drain(fixture.io);
  BOOST_REQUIRE_EQUAL(fixture.scores->requests.size(), 2U);
  BOOST_CHECK(!fixture.loader.snapshot().document);
  BOOST_CHECK_EQUAL(*fixture.completions, 0U);
  fixture.loader.select(fixture.catalog, std::nullopt);
  fixture.scores->complete(1);
  drain(fixture.io);
  BOOST_CHECK(!fixture.loader.snapshot().document);
  BOOST_CHECK(!fixture.loader.snapshot().loading);
  BOOST_CHECK_EQUAL(*fixture.completions, 0U);
}

BOOST_AUTO_TEST_CASE(score_loader_errors_persist_without_repeated_loads_and_allow_reselection) {
  auto fixture = loader_fixture{};
  fixture.scores->fails = true;
  fixture.loader.select(fixture.catalog, performance("missing"));
  drain(fixture.io);
  fixture.scores->complete(0);
  drain(fixture.io);
  auto loaded = fixture.loader.snapshot();
  BOOST_REQUIRE(loaded.error);
  BOOST_CHECK(loaded.error->describe().find("missing fixture score") != std::string::npos);
  BOOST_CHECK(!loaded.loading);
  BOOST_CHECK(!loaded.document);
  fixture.loader.select(fixture.catalog, performance("missing"));
  drain(fixture.io);
  BOOST_CHECK_EQUAL(fixture.scores->requests.size(), 1U);
  fixture.loader.select(fixture.catalog, std::nullopt);
  fixture.scores->fails = false;
  fixture.loader.select(fixture.catalog, performance("missing"));
  drain(fixture.io);
  fixture.scores->complete(1);
  drain(fixture.io);
  BOOST_REQUIRE(fixture.loader.snapshot().document);
  BOOST_CHECK(!fixture.loader.snapshot().error);
}

BOOST_AUTO_TEST_CASE(score_loader_converts_coroutine_exceptions_to_error_state) {
  auto fixture = loader_fixture{};
  fixture.scores->throws = true;
  fixture.loader.select(fixture.catalog, performance("a"));
  drain(fixture.io);
  fixture.scores->complete(0);
  drain(fixture.io);
  BOOST_CHECK(fixture.loader.snapshot().error.has_value());
  BOOST_CHECK(!fixture.loader.snapshot().loading);
  BOOST_CHECK_EQUAL(*fixture.completions, 1U);
}

BOOST_AUTO_TEST_CASE(score_loader_stop_discards_inflight_and_pending_loads) {
  auto fixture = loader_fixture{};
  fixture.loader.select(fixture.catalog, performance("a"));
  drain(fixture.io);
  fixture.loader.select(fixture.catalog, performance("b"));
  fixture.loader.stop();
  fixture.scores->complete(0);
  fixture.loader.select(fixture.catalog, performance("c"));
  drain(fixture.io);
  BOOST_CHECK_EQUAL(fixture.scores->requests.size(), 1U);
  BOOST_CHECK_EQUAL(*fixture.completions, 0U);
  BOOST_CHECK(!fixture.loader.snapshot().document);
  BOOST_CHECK(!fixture.loader.snapshot().loading);
}

BOOST_AUTO_TEST_CASE(score_loader_stop_before_executor_runs_does_not_start_load) {
  auto fixture = loader_fixture{};
  fixture.loader.select(fixture.catalog, performance("a"));
  fixture.loader.stop();
  drain(fixture.io);
  BOOST_CHECK(fixture.scores->requests.empty());
  BOOST_CHECK_EQUAL(*fixture.completions, 0U);
}

BOOST_AUTO_TEST_CASE(score_loader_destruction_during_load_does_not_call_consumer) {
  auto io = boost::asio::io_context{};
  auto scores = std::make_shared<controlled_scores>(io);
  auto completions = std::size_t{};
  {
    auto loader = vamp_zero::score_loader{io.get_executor(),
                                          [scores](vamp_zero::score_reference reference) { return scores->load(std::move(reference)); },
                                          [&] { ++completions; }};
    loader.select(std::make_shared<vamp_zero::catalog const>(), performance("a"));
    drain(io);
    BOOST_REQUIRE_EQUAL(scores->requests.size(), 1U);
  }
  scores->complete(0);
  drain(io);
  BOOST_CHECK_EQUAL(completions, 0U);
}

BOOST_AUTO_TEST_CASE(score_loader_completion_can_overlap_stop_and_consumer_destruction) {
  auto io = boost::asio::io_context{};
  auto entered = std::promise<void>{};
  auto entered_future = entered.get_future();
  auto resume = std::promise<void>{};
  auto resume_future = resume.get_future().share();
  auto consumer = std::make_shared<std::atomic<std::size_t>>(0U);
  auto loader = std::make_unique<vamp_zero::score_loader>(io.get_executor(), [](vamp_zero::score_reference reference) -> boost::asio::awaitable<std::expected<vamp_zero::score_document, modulus::error>> { co_return vamp_zero::score_document{.id = reference.id}; }, [weak = std::weak_ptr{consumer}, &entered, resume_future] {
      entered.set_value();
      resume_future.wait();
      if (auto target = weak.lock()) {
        ++*target;
      } });
  loader->select(std::make_shared<vamp_zero::catalog const>(), performance("a"));
  auto worker = std::thread{[&] { io.run(); }};
  auto entered_status = entered_future.wait_for(std::chrono::seconds{5});
  if (entered_status == std::future_status::ready) {
    auto document = std::weak_ptr{loader->snapshot().document};
    loader->stop();
    BOOST_CHECK(document.expired());
    loader.reset();
    consumer.reset();
  }
  resume.set_value();
  worker.join();
  BOOST_CHECK(entered_status == std::future_status::ready);
}

BOOST_AUTO_TEST_CASE(selected_score_reference_uses_active_part) {
  auto catalog = std::make_shared<vamp_zero::catalog>();
  catalog->parts = {{.id = "part-a"}, {.id = "part-b"}};
  auto state = vamp_zero::app_state{.catalog = catalog};
  BOOST_CHECK(!vamp_zero::selected_score_reference(state));
  state.mode = vamp_zero::app_mode::PERFORMANCE_SCORE;
  state.active_part_id = "part-b";
  auto reference = vamp_zero::selected_score_reference(state);
  BOOST_REQUIRE(reference);
  BOOST_CHECK_EQUAL(reference->id, "part-b");
  state.active_part_id = "missing";
  BOOST_CHECK(!vamp_zero::selected_score_reference(state));
  state.catalog.reset();
  BOOST_CHECK(!vamp_zero::selected_score_reference(state));
}