diff --git a/src/client/mcp_client.cc b/src/client/mcp_client.cc index 3593c11d2..cab760718 100644 --- a/src/client/mcp_client.cc +++ b/src/client/mcp_client.cc @@ -1010,6 +1010,28 @@ void McpClient::handleClientStreamEvent(ClientStreamEvent event, void McpClient::resumeAnswer(const std::shared_ptr& context, const std::string& last_event_id) { + if (streamable_session_ && protocol::modern::isModernVersion( + streamable_session_->protocolVersion())) { + // Nothing to pick it up with. This era numbers no events, so there + // is no place to carry on from, and it serves no standalone stream + // to carry on over — asking for one would be asking for a method + // this server refuses, and reading that refusal as a fact about + // streams rather than about this request would leave the request + // outstanding forever. + // + // So a stream cut short is the end of what it was carrying. Said + // rather than waited on, and said through the ordinary failure path, + // which is also what lets go of a subscription. + GOPHER_LOG_DEBUG("The stream carrying {} was cut and cannot be resumed", + context->method); + stream_recovering_.reset(); + completeRequestWithError( + context, Error(::mcp::jsonrpc::INTERNAL_ERROR, + "The stream carrying this answer was cut off, and this " + "revision has no way to pick one up again")); + return; + } + if (context->resume_attempts >= config_.streamable_http.resume_attempts) { GOPHER_LOG_WARN("Giving up on the answer to {} after {} attempts", context->method, context->resume_attempts); diff --git a/tests/integration/test_modern_era_subscriptions.cc b/tests/integration/test_modern_era_subscriptions.cc index 88b72ab58..a4b9b6b58 100644 --- a/tests/integration/test_modern_era_subscriptions.cc +++ b/tests/integration/test_modern_era_subscriptions.cc @@ -589,5 +589,64 @@ TEST_F(RefusedSubscriptionTest, EndingOneSeversNoOtherRequest) { "ending"; } +// The drop that comes from the far end rather than from here. Its +// request is still outstanding when the stream goes — which is what the +// local case never exercises, since ending one takes its request away +// before the connection closes. +// +// A stream cut short in this era is the end of what it was carrying: +// there are no event ids to carry on from and no standalone stream to +// carry on over, so asking for one asks for a method this server refuses +// — and reading that refusal as a fact about streams rather than about +// this request leaves the request outstanding for good. +TEST_F(RefusedSubscriptionTest, ADroppedSubscriptionIsEndedNotResumed) { + const uint16_t port = + server_.start([](const test::Seen& seen) -> test::Reply { + if (seen.rpc_method == modern::kMethodServerDiscover) { + return test::Reply::write(test::withBody( + 200, "OK", "application/json", + "{\"jsonrpc\":\"2.0\",\"id\":\"d\",\"result\":{\"resultType\":" + "\"complete\",\"supportedVersions\":[\"2026-07-28\"]," + "\"capabilities\":{}}}", + std::string())); + } + if (seen.rpc_method == modern::kMethodSubscriptionsListen) { + return test::Reply::stream(test::streamPrelude()); + } + return test::Reply::write(test::accepted()); + }); + + client::McpClientConfig config; + config.client_name = "dropped-subscription-client"; + config.client_version = "0.0.1"; + config.num_workers = 1; + config.request_timeout = 5000ms; + config.protocol_connection_timeout = 5000ms; + config.streamable_http.fallback_probe_timeout = 700ms; + client_ = client::createMcpClient(config); + ASSERT_NE(client_, nullptr); + ASSERT_TRUE(holds_alternative( + client_->connect("http://127.0.0.1:" + std::to_string(port) + "/mcp"))); + + modern::NotificationFilter what; + what.tools_list_changed = true; + const int64_t id = client_->listen(what, [](const jsonrpc::Notification&) {}); + ASSERT_NE(id, 0); + ASSERT_TRUE(server_.waitForRpc(modern::kMethodSubscriptionsListen, 1)); + ASSERT_TRUE(waitUntilTrue( + [this]() { return client_->subscriptionsHeld() == 1; }, 3000ms)); + std::this_thread::sleep_for(300ms); + + const size_t streams_before = server_.countOfMethod("GET"); + server_.cutStream(); + + EXPECT_TRUE(waitUntilTrue( + [this]() { return client_->subscriptionsHeld() == 0; }, 3000ms)) + << "a subscription whose stream was dropped is still being held"; + EXPECT_EQ(server_.countOfMethod("GET"), streams_before) + << "a dropped subscription was asked for again over a stream, in a " + "revision that serves none"; +} + } // namespace } // namespace mcp diff --git a/tests/interop/child_process.h b/tests/interop/child_process.h index ed563a3f3..7408b3485 100644 --- a/tests/interop/child_process.h +++ b/tests/interop/child_process.h @@ -13,12 +13,16 @@ #pragma once +#include #include #include +#include +#include #include #include #include #include +#include #include #include @@ -30,17 +34,43 @@ namespace mcp { namespace test { /** A loopback port the kernel believes is free. */ -inline uint16_t pickFreePort() { +inline uint16_t pickFreePort(std::string* why_not = nullptr) { + auto say = [why_not](const std::string& reason) { + if (why_not != nullptr) { + *why_not = reason; + } + }; + auto& iface = network::socketInterface(); auto fd = iface.socket(network::SocketType::Stream, network::Address::Type::Ip, network::Address::IpVersion::v4); if (!fd.ok()) { + say("no socket could be made to find a free port with"); return 0; } auto handle = iface.ioHandleForFd(*fd, false); - handle->bind(network::Address::parseInternetAddress("127.0.0.1", 0)); - handle->listen(1); + + // Checked, unlike before. A bind or listen that fails leaves the port + // at zero and the caller with nothing to go on — and on a machine + // where this is what is wrong, that is the whole of what it needs to + // be told. + auto bound = + handle->bind(network::Address::parseInternetAddress("127.0.0.1", 0)); + if (!bound.ok()) { + say("binding a loopback port failed (errno " + std::to_string(errno) + + "); the loopback interface may be down or restricted"); + handle->close(); + return 0; + } + auto listening = handle->listen(1); + if (!listening.ok()) { + say("listening on a loopback port failed (errno " + std::to_string(errno) + + ")"); + handle->close(); + return 0; + } + auto local = handle->localAddress(); uint16_t port = 0; if (local.ok()) { @@ -49,10 +79,69 @@ inline uint16_t pickFreePort() { port = ip->port(); } } + if (port == 0) { + say("a loopback port was bound but the kernel did not name it"); + } handle->close(); return port; } +/** What `node --version` says, as major and minor. Zeroes when unknown. */ +inline std::pair nodeVersion() { + FILE* pipe = popen("node --version 2>/dev/null", "r"); + if (pipe == nullptr) { + return std::make_pair(0, 0); + } + char buffer[64] = {0}; + const char* got = fgets(buffer, sizeof(buffer), pipe); + pclose(pipe); + if (got == nullptr) { + return std::make_pair(0, 0); + } + std::string text(buffer); + if (!text.empty() && text[0] == 'v') { + text.erase(0, 1); + } + int major = 0; + int minor = 0; + if (std::sscanf(text.c_str(), "%d.%d", &major, &minor) != 2) { + return std::make_pair(0, 0); + } + return std::make_pair(major, minor); +} + +/** + * What this node needs in order to run a TypeScript file directly. + * + * Both interop programs are `.ts` run straight by node, which is only + * something node can do at all from 22.6, and only without being asked + * from 23.6. In between it has to be asked — and a node that is not + * asked fails immediately with a syntax error, which from the outside + * looks exactly like a server that started and never listened. + * + * @return False when this node cannot run them however it is asked. + */ +inline bool nodeTypeScriptFlags(std::vector* flags, + std::string* why_not) { + const auto version = nodeVersion(); + if (version.first == 0) { + *why_not = "node is installed but did not say what version it is"; + return false; + } + if (version.first < 22 || (version.first == 22 && version.second < 6)) { + *why_not = "node " + std::to_string(version.first) + "." + + std::to_string(version.second) + + " cannot run TypeScript directly; these programs need 22.6 " + "or newer"; + return false; + } + if (version.first < 23 || (version.first == 23 && version.second < 6)) { + // Able, but only when asked. + flags->push_back("--experimental-strip-types"); + } + return true; +} + /** True once something is accepting on the port, or the budget is spent. */ inline bool waitUntilAccepting(uint16_t port, std::chrono::milliseconds budget) { @@ -147,47 +236,101 @@ class Child { return true; } + /** + * How a child ended, for a caller that has to say. + * + * Three outcomes and not two: a child that was killed is not one that + * exited with a code, and neither is one that is still going. Told + * apart because the caller that asks is explaining why something did + * not start, and "it crashed" is the answer it is most often looking + * for. + */ + struct Ending { + enum class How { StillRunning, Exited, Signalled }; + + How how{How::StillRunning}; + /** The exit status when it exited, the signal number when it was + * killed, and nothing meaningful while it is still going. */ + int code{-1}; + + bool exitedWith(int expected) const { + return how == How::Exited && code == expected; + } + + std::string describe() const { + if (how == How::Exited) { + return "exited with code " + std::to_string(code); + } + if (how == How::Signalled) { + const char* named = strsignal(code); + return "was killed by signal " + std::to_string(code) + + (named != nullptr ? std::string(" (") + named + ")" + : std::string()); + } + return "was still running"; + } + }; + /** * Wait for it to finish, keeping whatever it wrote. * - * The pipe is drained to EOF first: reaping a child whose output has not - * been read is how a test ends up reporting a failure with nothing to - * say about it. + * Both at once, and neither before the other. Draining to EOF first + * looked like the tidy order — read everything, then reap — but EOF on + * a child's pipe means every write end has closed, which for a child + * that is still running is a thing that has not happened yet. Waiting + * for it there is waiting for the child to exit, without a deadline, + * inside the call whose whole purpose is to have one. A caller asking + * why a peer that is still running has not started would have hung on + * the question. * - * @return Its exit status, or -1 if it did not finish in time. + * So the pipe is read for whatever is there, the child is reaped if it + * has gone, and the budget is checked — round and round until one of + * the last two settles it. + * + * @return How it ended, and with what. What it wrote up to that point + * is kept whichever way that is — including when it is still + * running, which is what a caller under a deadline gets. */ - int wait(std::chrono::milliseconds budget) { + Ending wait(std::chrono::milliseconds budget) { + Ending ending; if (pid_ <= 0) { - return -1; - } - if (read_fd_ >= 0) { - char buffer[4096]; - ssize_t got = 0; - while ((got = read(read_fd_, buffer, sizeof(buffer))) > 0) { - output_.append(buffer, static_cast(got)); - } - close(read_fd_); - read_fd_ = -1; + return ending; } const auto deadline = std::chrono::steady_clock::now() + budget; int status = 0; - while (std::chrono::steady_clock::now() < deadline) { + for (;;) { + drainWhatIsThere(); + const pid_t done = waitpid(pid_, &status, WNOHANG); if (done == pid_) { + // Gone, so its pipe is at EOF and the rest of what it wrote is + // there to be had. + drainWhatIsThere(); + closeRead(); pid_ = -1; - return WIFEXITED(status) ? WEXITSTATUS(status) : -1; + if (WIFEXITED(status)) { + ending.how = Ending::How::Exited; + ending.code = WEXITSTATUS(status); + } else if (WIFSIGNALED(status)) { + ending.how = Ending::How::Signalled; + ending.code = WTERMSIG(status); + } + return ending; + } + + if (std::chrono::steady_clock::now() >= deadline) { + return ending; } std::this_thread::sleep_for(std::chrono::milliseconds(20)); } - return -1; } void stop() { - if (read_fd_ >= 0) { - close(read_fd_); - read_fd_ = -1; - } + // Whatever it managed to say before being told to go, since a child + // that has to be killed is one somebody will want an account of. + drainWhatIsThere(); + closeRead(); if (pid_ <= 0) { return; } @@ -207,6 +350,43 @@ class Child { bool running() const { return pid_ > 0; } + private: + /** Take what has been written so far, and never wait for more. */ + void drainWhatIsThere() { + if (read_fd_ < 0) { + return; + } + for (;;) { + struct pollfd waiting; + waiting.fd = read_fd_; + waiting.events = POLLIN; + waiting.revents = 0; + // No timeout at all: this asks what is there, never what is + // coming. + const int ready = poll(&waiting, 1, 0); + if (ready <= 0) { + return; + } + char buffer[4096]; + const ssize_t got = read(read_fd_, buffer, sizeof(buffer)); + if (got > 0) { + output_.append(buffer, static_cast(got)); + continue; + } + // Zero is EOF and negative is nothing more to be had now; either + // way there is nothing further to read in this pass. + return; + } + } + + void closeRead() { + if (read_fd_ >= 0) { + close(read_fd_); + read_fd_ = -1; + } + } + + public: /** Everything it wrote, once wait() has read it. */ const std::string& output() const { return output_; } diff --git a/tests/interop/test_client_vs_official_server.cc b/tests/interop/test_client_vs_official_server.cc index 92173ccb8..a13d2c902 100644 --- a/tests/interop/test_client_vs_official_server.cc +++ b/tests/interop/test_client_vs_official_server.cc @@ -40,6 +40,7 @@ namespace { using namespace std::chrono_literals; using test::Child; +using test::nodeTypeScriptFlags; using test::pickFreePort; using test::waitUntilAccepting; @@ -81,6 +82,13 @@ bool referenceServerAvailable(std::string& why_not) { dir; return false; } + // Asked here rather than left to fail later: a node that cannot run + // TypeScript exits the moment it is handed some, which looks from the + // outside like a server that started and never listened. + std::vector unused; + if (!nodeTypeScriptFlags(&unused, &why_not)) { + return false; + } return true; } @@ -88,20 +96,54 @@ bool referenceServerAvailable(std::string& why_not) { class ReferenceServer { public: /** Start it, and wait until something is accepting on its port. */ - bool start(const std::vector& flags = {}) { - port_ = pickFreePort(); + ::testing::AssertionResult start(const std::vector& flags = {}) { + std::string why_not; + port_ = pickFreePort(&why_not); if (port_ == 0) { - return false; + return ::testing::AssertionFailure() + << "no port for the reference server: " << why_not; + } + + std::vector argv{"node"}; + if (!nodeTypeScriptFlags(&argv, &why_not)) { + return ::testing::AssertionFailure() + << "the reference server cannot be run: " << why_not; } - std::vector argv{"node", "server.ts", "--port", - std::to_string(port_)}; + argv.push_back("server.ts"); + argv.push_back("--port"); + argv.push_back(std::to_string(port_)); for (const auto& flag : flags) { argv.push_back(flag); } - if (!process_.start(referenceServerDir(), argv)) { - return false; + + // Kept, unlike before. What it wrote is the only account of why it + // did not come up, and discarding it left a ten-second silence and a + // bare false — which is not something anyone can act on without + // reproducing the whole thing by hand. + if (!process_.start(referenceServerDir(), argv, /*capture=*/true)) { + return ::testing::AssertionFailure() + << "the reference server could not be started in " + << referenceServerDir(); + } + if (waitUntilAccepting(port_, 10s)) { + return ::testing::AssertionSuccess(); } - return waitUntilAccepting(port_, 10s); + + // Still running and not listening, exited, or killed. All three are + // worth telling apart — a crash on startup is the likeliest of them, + // and the one a caller most wants named — and all three are + // explained by what it wrote. + // + // Asked of the reaping rather than looked at before it: a child that + // has already died is still running as far as anything that has not + // reaped it is concerned. + const Child::Ending ending = process_.wait(1s); + const std::string said = process_.output(); + return ::testing::AssertionFailure() + << "the reference server did not accept on port " << port_ + << " within 10s; it " << ending.describe() + << (said.empty() ? std::string(" and wrote nothing") + : std::string(" and wrote:\n") + said); } void stop() { process_.stop(); } diff --git a/tests/interop/test_official_client_vs_server.cc b/tests/interop/test_official_client_vs_server.cc index 81bc91ea6..2dd665417 100644 --- a/tests/interop/test_official_client_vs_server.cc +++ b/tests/interop/test_official_client_vs_server.cc @@ -47,6 +47,7 @@ namespace { using namespace std::chrono_literals; using test::Child; +using test::nodeTypeScriptFlags; using test::pickFreePort; using test::waitUntilAccepting; @@ -97,6 +98,14 @@ bool driverAvailable(std::string& why_not) { "the driver's dependencies are not installed; run `npm ci` in " + dir; return false; } + // Asked here, with everything else this needs before it can run at + // all. Left to the run itself, a node too old to read the driver is + // four failures that all mean "this machine cannot host the test" — + // which is what skipping says, and says once. + std::vector unused; + if (!nodeTypeScriptFlags(&unused, &why_not)) { + return false; + } if (!fileExists(serverBinary())) { why_not = "the interop server is not at " + serverBinary() + "; build the gopher_interop_server target"; @@ -115,23 +124,31 @@ bool driverAvailable(std::string& why_not) { */ struct DriverRun { int status{-1}; + /** How the driver ended, in words, so a crash reads as one. */ + std::string ending{"never ran"}; std::string output; }; DriverRun driveServer(const std::vector& modes) { DriverRun run; - const uint16_t port = pickFreePort(); + std::string why_not; + const uint16_t port = pickFreePort(&why_not); if (port == 0) { - run.output = "no free port"; + run.output = "no free port for the interop run: " + why_not; return run; } std::vector server_argv{serverBinary(), "--port", std::to_string(port)}; - std::vector driver_argv{ - "node", "client.ts", "--url", - "http://127.0.0.1:" + std::to_string(port) + "/mcp"}; + std::vector driver_argv{"node"}; + if (!nodeTypeScriptFlags(&driver_argv, &why_not)) { + run.output = "the driver cannot be run: " + why_not; + return run; + } + driver_argv.push_back("client.ts"); + driver_argv.push_back("--url"); + driver_argv.push_back("http://127.0.0.1:" + std::to_string(port) + "/mcp"); for (const auto& mode : modes) { server_argv.push_back(mode); // --no-resume changes what the server keeps, not what the client may @@ -142,12 +159,21 @@ DriverRun driveServer(const std::vector& modes) { } Child server; - if (!server.start(std::string(), server_argv)) { + if (!server.start(std::string(), server_argv, /*capture=*/true)) { run.output = "could not start " + serverBinary(); return run; } if (!waitUntilAccepting(port, 10s)) { - run.output = "the server never accepted a connection"; + // Whatever it wrote on its way to not listening, which is the only + // account of why there is nothing to talk to — and how it ended, + // taken from the reaping rather than from a look before it, since a + // child that has already died is still running as far as anything + // that has not reaped it is concerned. + const Child::Ending ending = server.wait(1s); + run.output = serverBinary() + " never accepted on port " + + std::to_string(port) + "; it " + ending.describe() + + (server.output().empty() ? " and wrote nothing" + : " and wrote:\n" + server.output()); return run; } @@ -157,14 +183,17 @@ DriverRun driveServer(const std::vector& modes) { return run; } - run.status = driver.wait(120s); + const Child::Ending ending = driver.wait(120s); + run.status = ending.how == Child::Ending::How::Exited ? ending.code : -1; + run.ending = ending.describe(); run.output = driver.output(); return run; } /** Fails with the driver's own report rather than a status code. */ void expectClean(const DriverRun& run) { - EXPECT_EQ(run.status, 0) << "the driver reported a failure:\n" << run.output; + EXPECT_EQ(run.status, 0) << "the driver " << run.ending << ":\n" + << run.output; if (run.status != 0) { return; } @@ -216,5 +245,81 @@ TEST_F(OfficialClientVsServer, AServerRetainingNothingIsStillServed) { expectClean(run); } +// ===== The harness itself ===== +// +// Not about either implementation: about the thing both directions use +// to run one. It needs no node, so it runs wherever the suite is built. + +// Asking a still-running child how it is going has to come back. Its +// pipe is only at EOF once every write end has closed, which for a child +// that is still running has not happened — so draining to EOF first is +// waiting for it to exit, inside the call whose whole purpose is not to +// wait longer than it was told. +// +// This is the case the diagnostics reach for by design: they ask exactly +// when a peer is still running and has not started serving. +TEST(InteropHarness, AskingAStillRunningChildComesBack) { + test::Child child; + // Says something, so there is output to drain, then stays — with its + // pipe open, which is what makes the difference. + ASSERT_TRUE(child.start(std::string(), + {"/bin/sh", "-c", "echo listening; sleep 30"}, + /*capture=*/true)); + + const auto began = std::chrono::steady_clock::now(); + const Child::Ending ending = child.wait(500ms); + const auto took = std::chrono::steady_clock::now() - began; + + EXPECT_EQ(ending.how, Child::Ending::How::StillRunning) + << "a child that is still running was reported as done: " + << ending.describe(); + EXPECT_LT(took, 5s) + << "asking about a still-running child waited for it to exit instead " + "of for the time it was given"; + + // And what it said before it settled is still to be had, which is the + // whole reason for draining at all. + EXPECT_NE(child.output().find("listening"), std::string::npos) + << "what the child wrote was lost: '" << child.output() << "'"; + + child.stop(); +} + +// A child that ends is still reaped and read to the end. +TEST(InteropHarness, AChildThatEndsIsReadToTheEnd) { + test::Child child; + ASSERT_TRUE(child.start(std::string(), {"/bin/sh", "-c", "echo done; exit 3"}, + /*capture=*/true)); + + const Child::Ending ending = child.wait(5s); + EXPECT_TRUE(ending.exitedWith(3)) << "it " << ending.describe(); + EXPECT_NE(child.output().find("done"), std::string::npos) + << "what the child wrote was lost: '" << child.output() << "'"; +} + +// A peer that crashed on startup is the case the diagnostics exist for, +// and the one most easily got wrong: it is still running as far as +// anything that has not reaped it is concerned, so an answer taken +// before the reaping says "still running" about a child that is long +// dead — and says it exactly when a crash is what happened. +TEST(InteropHarness, AChildKilledBySignalIsNotReportedAsRunning) { + test::Child child; + ASSERT_TRUE(child.start(std::string(), + {"/bin/sh", "-c", "echo starting; kill -SEGV $$"}, + /*capture=*/true)); + + const Child::Ending ending = child.wait(5s); + EXPECT_EQ(ending.how, Child::Ending::How::Signalled) + << "a child that was killed was reported as having " << ending.describe(); + EXPECT_EQ(ending.code, SIGSEGV); + EXPECT_NE(ending.describe().find("killed by signal"), std::string::npos) + << ending.describe(); + + // And what it managed to say first is still to be had. + EXPECT_NE(child.output().find("starting"), std::string::npos) + << "what the child wrote before it died was lost: '" << child.output() + << "'"; +} + } // namespace } // namespace mcp