// AsioTransportAdapter unit tests for maxInflight exhaustion behavior #include #include #include #include #include #include #include #include #include #include #include #include #include #ifndef _WIN32 #include #endif using namespace yams::daemon; using yams::ErrorCode; using namespace std::chrono_literals; namespace fs = std::filesystem; namespace { fs::path makeTempRuntimeDir(const std::string& name) { auto base = fs::temp_directory_path(); #ifndef _WIN32 constexpr std::size_t maxUnixPath = sizeof(sockaddr_un::sun_path) - 1; auto candidate = base / "yams-transport-tests" / name / "ipc.sock"; if (candidate.native().size() >= maxUnixPath) { base = fs::path("/tmp"); } #endif auto dir = base / "yams-transport-tests" / name; std::error_code ec; fs::create_directories(dir, ec); return dir; } std::string randomSuffix() { return std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()); } } // namespace TEST_CASE("AsioTransportAdapter returns ResourceExhausted when maxInflight exceeded", "[daemon][transport][unit]") { #ifdef _WIN32 SKIP("Unix domain socket tests skipped on Windows"); #endif auto runtimeDir = makeTempRuntimeDir("max-inflight-" + randomSuffix()); auto socketPath = runtimeDir / "ipc.sock"; std::error_code ec; fs::remove(socketPath, ec); boost::asio::io_context server_io; boost::asio::local::stream_protocol::acceptor acceptor( server_io, boost::asio::local::stream_protocol::endpoint(socketPath.string())); std::atomic stop{false}; std::promise accepted; auto acceptedFuture = accepted.get_future(); std::thread server_thread([&] { try { boost::system::error_code bec; boost::asio::local::stream_protocol::socket sock(server_io); acceptor.accept(sock, bec); if (bec) { accepted.set_exception( std::make_exception_ptr(std::runtime_error("accept failed: " + bec.message()))); return; } // Only allow a single client connection. This forces all client requests to reuse the // same underlying socket, so maxInflight is exercised on one connection. acceptor.close(bec); accepted.set_value(); while (!stop.load(std::memory_order_relaxed)) { std::this_thread::sleep_for(5ms); } sock.close(bec); } catch (...) { try { accepted.set_exception(std::current_exception()); } catch (...) { } } }); TransportOptions opts; opts.socketPath = socketPath; opts.poolEnabled = true; opts.maxInflight = 2; opts.requestTimeout = 10s; opts.headerTimeout = 60s; opts.bodyTimeout = 60s; AsioTransportAdapter adapter(opts); // Launch two requests that will occupy the in-flight slots (server never responds). auto makeReq = []() -> Request { return Request{StatusRequest{}}; }; auto fut1 = boost::asio::co_spawn(GlobalIOContext::global_executor(), adapter.send_request(makeReq()), boost::asio::use_future); REQUIRE(acceptedFuture.wait_for(2s) == std::future_status::ready); REQUIRE_NOTHROW(acceptedFuture.get()); // Give request #1 time to connect + write and release the connection for reuse. std::this_thread::sleep_for(50ms); auto fut2 = boost::asio::co_spawn(GlobalIOContext::global_executor(), adapter.send_request(makeReq()), boost::asio::use_future); std::this_thread::sleep_for(50ms); // Third request should fail immediately with ResourceExhausted due to maxInflight. auto fut3 = boost::asio::co_spawn(GlobalIOContext::global_executor(), adapter.send_request(makeReq()), boost::asio::use_future); REQUIRE(fut3.wait_for(1s) == std::future_status::ready); auto r3 = fut3.get(); REQUIRE_FALSE(r3.has_value()); REQUIRE(r3.error().code == ErrorCode::ResourceExhausted); // Unblock the first two requests by closing the server-side socket. stop.store(true, std::memory_order_relaxed); server_thread.join(); REQUIRE(fut1.wait_for(2s) == std::future_status::ready); REQUIRE(fut2.wait_for(2s) == std::future_status::ready); auto r1 = fut1.get(); auto r2 = fut2.get(); REQUIRE_FALSE(r1.has_value()); REQUIRE_FALSE(r2.has_value()); fs::remove(socketPath, ec); AsioConnectionPool::shutdown_all(100ms); }