253 lines
9.1 KiB
C++
253 lines
9.1 KiB
C++
#if __has_include(<catch2/catch_test_macros.hpp>)
|
|
#include <catch2/catch_test_macros.hpp>
|
|
#elif __has_include("catch2/catch_test_macros.hpp")
|
|
#include "catch2/catch_test_macros.hpp"
|
|
#endif
|
|
|
|
#include <algorithm>
|
|
#include <cstdlib>
|
|
#include <filesystem>
|
|
#include <fstream>
|
|
#include <mutex>
|
|
#include <string>
|
|
#include <vector>
|
|
|
|
#if __has_include(<boost/asio/co_spawn.hpp>)
|
|
#include <boost/asio/co_spawn.hpp>
|
|
#elif __has_include("boost/asio/co_spawn.hpp")
|
|
#include "boost/asio/co_spawn.hpp"
|
|
#endif
|
|
#if __has_include(<boost/asio/io_context.hpp>)
|
|
#include <boost/asio/io_context.hpp>
|
|
#elif __has_include("boost/asio/io_context.hpp")
|
|
#include "boost/asio/io_context.hpp"
|
|
#endif
|
|
#if __has_include(<boost/asio/use_future.hpp>)
|
|
#include <boost/asio/use_future.hpp>
|
|
#elif __has_include("boost/asio/use_future.hpp")
|
|
#include "boost/asio/use_future.hpp"
|
|
#endif
|
|
|
|
#if __has_include(<yams/daemon/client/in_process_transport.h>)
|
|
#include <yams/daemon/client/in_process_transport.h>
|
|
#elif __has_include("yams/daemon/client/in_process_transport.h")
|
|
#include "yams/daemon/client/in_process_transport.h"
|
|
#else
|
|
#include "../../../include/yams/daemon/client/in_process_transport.h"
|
|
#endif
|
|
#if __has_include(<yams/daemon/embedded_service_host.h>)
|
|
#include <yams/daemon/embedded_service_host.h>
|
|
#elif __has_include("yams/daemon/embedded_service_host.h")
|
|
#include "yams/daemon/embedded_service_host.h"
|
|
#else
|
|
#include "../../../include/yams/daemon/embedded_service_host.h"
|
|
#endif
|
|
|
|
#include "../../common/test_helpers_catch2.h"
|
|
|
|
#if defined(__has_feature)
|
|
#if __has_feature(undefined_behavior_sanitizer)
|
|
#define YAMS_UBSAN_ACTIVE 1
|
|
#endif
|
|
#endif
|
|
#if defined(__SANITIZE_UNDEFINED__)
|
|
#define YAMS_UBSAN_ACTIVE 1
|
|
#endif
|
|
#if defined(__has_feature)
|
|
#if __has_feature(thread_sanitizer)
|
|
#define YAMS_TSAN_ACTIVE 1
|
|
#endif
|
|
#endif
|
|
#if defined(__SANITIZE_THREAD__)
|
|
#define YAMS_TSAN_ACTIVE 1
|
|
#endif
|
|
|
|
namespace fs = std::filesystem;
|
|
|
|
namespace {
|
|
|
|
yams::Result<void> run_streaming(yams::daemon::InProcessTransport& transport,
|
|
yams::daemon::Request request,
|
|
yams::daemon::IClientTransport::HeaderCallback onHeader,
|
|
yams::daemon::IClientTransport::ChunkCallback onChunk,
|
|
yams::daemon::IClientTransport::ErrorCallback onError,
|
|
yams::daemon::IClientTransport::CompleteCallback onComplete) {
|
|
boost::asio::io_context io;
|
|
auto headerCallback = std::move(onHeader);
|
|
auto chunkCallback = std::move(onChunk);
|
|
auto errorCallback = std::move(onError);
|
|
auto completeCallback = std::move(onComplete);
|
|
auto fut = boost::asio::co_spawn(
|
|
io,
|
|
transport.send_request_streaming(std::move(request), headerCallback, chunkCallback,
|
|
errorCallback, completeCallback),
|
|
boost::asio::use_future);
|
|
io.run();
|
|
return fut.get();
|
|
}
|
|
|
|
std::shared_ptr<yams::daemon::EmbeddedServiceHost> make_host_for_test(const fs::path& dataDir) {
|
|
yams::daemon::EmbeddedServiceHost::Options options;
|
|
options.dataDir = dataDir;
|
|
options.ioThreads = 1;
|
|
options.enableAutoRepair = false;
|
|
options.autoLoadPlugins = false;
|
|
options.enableModelProvider = false;
|
|
options.initTimeoutSeconds = 30;
|
|
|
|
auto host = yams::daemon::EmbeddedServiceHost::getOrCreate(options);
|
|
REQUIRE((host.has_value()));
|
|
return host.value();
|
|
}
|
|
|
|
} // namespace
|
|
|
|
TEST_CASE("InProcessTransport orders callbacks for stream-capable requests",
|
|
"[daemon][streaming][inprocess]") {
|
|
#ifdef _WIN32
|
|
SKIP("InProcessTransport embedded host tests are unstable on Windows");
|
|
#endif
|
|
#ifdef YAMS_UBSAN_ACTIVE
|
|
SKIP("InProcessTransport embedded host teardown trips a known UBSan timer-service race");
|
|
#endif
|
|
#ifdef YAMS_TSAN_ACTIVE
|
|
SKIP("TSAN-only crash in stream-capable in-process transport path");
|
|
#endif
|
|
const auto dataDir = yams::test::make_temp_dir("yams_in_process_order_stream_");
|
|
|
|
yams::test::ScopedEnvVar dataEnv("YAMS_DATA_DIR", dataDir.string());
|
|
yams::test::ScopedEnvVar storageEnv("YAMS_STORAGE", dataDir.string());
|
|
yams::test::ScopedEnvVar disableVectors("YAMS_DISABLE_VECTORS", std::string("1"));
|
|
yams::test::ScopedEnvVar skipModelLoading("YAMS_SKIP_MODEL_LOADING", std::string("1"));
|
|
yams::test::ScopedEnvVar disableWatcher("YAMS_DISABLE_SESSION_WATCHER", std::string("1"));
|
|
yams::test::ScopedEnvVar syncSingleFileAdd("YAMS_SYNC_SINGLE_FILE_ADD", std::string("1"));
|
|
|
|
const auto inputPath = dataDir / "stream-order.txt";
|
|
{
|
|
std::ofstream out(inputPath);
|
|
REQUIRE((out.is_open()));
|
|
out << "streaming request body\n";
|
|
}
|
|
|
|
auto host = make_host_for_test(dataDir);
|
|
yams::daemon::InProcessTransport transport(host);
|
|
|
|
std::mutex eventsMutex;
|
|
std::vector<std::string> events;
|
|
const auto recordEvent = [&](std::string event) {
|
|
std::lock_guard<std::mutex> lock(eventsMutex);
|
|
events.push_back(std::move(event));
|
|
};
|
|
yams::daemon::EmbeddedServiceHost::TestingShutdownSnapshot beforeJoin{};
|
|
bool sawBeforeJoinSnapshot = false;
|
|
host->testing_setShutdownHook(
|
|
[&](const yams::daemon::EmbeddedServiceHost::TestingShutdownSnapshot& snapshot) {
|
|
if (snapshot.phase ==
|
|
yams::daemon::EmbeddedServiceHost::TestingShutdownPhase::BeforeThreadJoin) {
|
|
beforeJoin = snapshot;
|
|
sawBeforeJoinSnapshot = true;
|
|
}
|
|
});
|
|
yams::daemon::AddDocumentRequest addReq;
|
|
addReq.path = inputPath.string();
|
|
addReq.noEmbeddings = true;
|
|
addReq.waitForProcessing = true;
|
|
yams::daemon::Request req = addReq;
|
|
|
|
auto result = run_streaming(
|
|
transport, req, [&](const yams::daemon::Response&) { recordEvent("header"); },
|
|
[&](const yams::daemon::Response&, bool) {
|
|
recordEvent("chunk");
|
|
return true;
|
|
},
|
|
[&](const yams::Error&) { recordEvent("error"); }, [&]() { recordEvent("complete"); });
|
|
|
|
std::vector<std::string> recordedEvents;
|
|
{
|
|
std::lock_guard<std::mutex> lock(eventsMutex);
|
|
recordedEvents = events;
|
|
}
|
|
|
|
REQUIRE((result.has_value()));
|
|
REQUIRE_FALSE(recordedEvents.empty());
|
|
REQUIRE((recordedEvents.front() == "header"));
|
|
REQUIRE((recordedEvents.back() == "complete"));
|
|
|
|
const auto completePos = std::find(recordedEvents.begin(), recordedEvents.end(), "complete");
|
|
REQUIRE((completePos != recordedEvents.end()));
|
|
REQUIRE((std::find(recordedEvents.begin(), completePos, "chunk") != completePos));
|
|
REQUIRE(
|
|
(std::find(recordedEvents.begin(), recordedEvents.end(), "error") == recordedEvents.end()));
|
|
|
|
auto shutdownResult = host->shutdown();
|
|
REQUIRE((shutdownResult.has_value()));
|
|
|
|
REQUIRE((sawBeforeJoinSnapshot));
|
|
CHECK((beforeJoin.ioContextAlive));
|
|
CHECK((beforeJoin.lifecycleAlive));
|
|
CHECK((beforeJoin.dispatcherAlive));
|
|
CHECK((beforeJoin.serviceManagerAlive));
|
|
CHECK((beforeJoin.ioThreadCount >= 1));
|
|
|
|
std::error_code ec;
|
|
fs::remove_all(dataDir, ec);
|
|
}
|
|
|
|
TEST_CASE("InProcessTransport orders callbacks for unary requests",
|
|
"[daemon][streaming][inprocess]") {
|
|
#ifdef _WIN32
|
|
SKIP("InProcessTransport embedded host tests are unstable on Windows");
|
|
#endif
|
|
#ifdef YAMS_UBSAN_ACTIVE
|
|
SKIP("InProcessTransport embedded host teardown trips a known UBSan timer-service race");
|
|
#endif
|
|
const auto dataDir = yams::test::make_temp_dir("yams_in_process_order_unary_");
|
|
|
|
yams::test::ScopedEnvVar dataEnv("YAMS_DATA_DIR", dataDir.string());
|
|
yams::test::ScopedEnvVar storageEnv("YAMS_STORAGE", dataDir.string());
|
|
yams::test::ScopedEnvVar disableVectors("YAMS_DISABLE_VECTORS", std::string("1"));
|
|
yams::test::ScopedEnvVar skipModelLoading("YAMS_SKIP_MODEL_LOADING", std::string("1"));
|
|
yams::test::ScopedEnvVar disableWatcher("YAMS_DISABLE_SESSION_WATCHER", std::string("1"));
|
|
|
|
auto host = make_host_for_test(dataDir);
|
|
yams::daemon::InProcessTransport transport(host);
|
|
|
|
std::mutex eventsMutex;
|
|
std::vector<std::string> events;
|
|
const auto recordEvent = [&](std::string event) {
|
|
std::lock_guard<std::mutex> lock(eventsMutex);
|
|
events.push_back(std::move(event));
|
|
};
|
|
yams::daemon::Request req = yams::daemon::StatusRequest{};
|
|
SECTION("Control response is delivered once") {}
|
|
SECTION("Non-control immediate error response is delivered once") {
|
|
yams::daemon::GetRequest get;
|
|
get.hash = std::string(64, 'f');
|
|
req = std::move(get);
|
|
}
|
|
|
|
auto result = run_streaming(
|
|
transport, req, [&](const yams::daemon::Response&) { recordEvent("header"); },
|
|
[&](const yams::daemon::Response&, bool) {
|
|
recordEvent("chunk");
|
|
return true;
|
|
},
|
|
[&](const yams::Error&) { recordEvent("error"); }, [&]() { recordEvent("complete"); });
|
|
|
|
std::vector<std::string> recordedEvents;
|
|
{
|
|
std::lock_guard<std::mutex> lock(eventsMutex);
|
|
recordedEvents = events;
|
|
}
|
|
|
|
REQUIRE((result.has_value()));
|
|
REQUIRE((recordedEvents.size() == 2));
|
|
REQUIRE((recordedEvents[0] == "header"));
|
|
REQUIRE((recordedEvents[1] == "complete"));
|
|
|
|
auto shutdownResult = host->shutdown();
|
|
REQUIRE((shutdownResult.has_value()));
|
|
|
|
std::error_code ec;
|
|
fs::remove_all(dataDir, ec);
|
|
}
|