feat: Support restart command and 204-halt in SSE contract test service

contract-tests/sse-contract-tests/include/entity_manager.hpp

@@ -1,24 +1,28 @@
 #pragma once
 
 #include <launchdarkly/logging/logger.hpp>
 #include "definitions.hpp"
 
 #include <launchdarkly/sse/client.hpp>
 
 #include <boost/asio/any_io_executor.hpp>
 
+#include <atomic>
 #include <memory>
 #include <mutex>
 #include <optional>
 #include <string>
+#include <tuple>
 #include <unordered_map>
 
 class EventOutbox;
 
 class EntityManager {
     using Inbox = std::shared_ptr<launchdarkly::sse::Client>;
     using Outbox = std::shared_ptr<EventOutbox>;
-    using Entity = std::pair<Inbox, Outbox>;
+    // Third element: suppress-next-error flag for caller-triggered restarts
+    // (the canceled read surfaces as an error on the read path).
+    using Entity = std::tuple<Inbox, Outbox, std::shared_ptr<std::atomic_bool>>;
 
     std::unordered_map<std::string, Entity> entities_;
 
@@ -49,4 +53,15 @@
      * @return True if the entity was found and destroyed.
      */
     bool destroy(std::string const& id);
+    /**
+     * Send a command to an entity (see the SSE contract test protocol).
+     * Currently supports "restart", which makes the client disconnect and
+     * reconnect immediately.
+     * @param id ID of the entity.
+     * @param command The command name.
+     * @return nullopt if no such entity exists; otherwise true if the
+     * command was recognized.
+     */
+    std::optional<bool> command(std::string const& id,
+                                std::string const& command);
 };

contract-tests/sse-contract-tests/src/entity_manager.cpp

@@ -1,3 +1,3 @@
 #include "entity_manager.hpp"
 #include "event_outbox.hpp"
 
@@ -44,13 +44,23 @@
         LD_LOG(logger_, LogLevel::kDebug) << std::move(msg);
     });
 
+    auto suppress_next_error = std::make_shared<std::atomic_bool>(false);
+
     client_builder.receiver([copy = poster](launchdarkly::sse::Event event) {
         copy->post_event(std::move(event));
     });
 
-    client_builder.errors([copy = poster](launchdarkly::sse::Error event) {
-        copy->post_error(event);
-    });
+    client_builder.errors(
+        [copy = poster, suppress = suppress_next_error](launchdarkly::sse::Error event) {
+            // A caller-triggered restart surfaces as an error on the read
+            // path (the canceled read looks like a body timeout); like other
+            // services (e.g. okhttp's StreamClosedByCallerException
+            // suppression) we don't report it to the harness.
+            if (suppress->exchange(false)) {
+                return;
+            }
+            copy->post_error(event);
+        });
 
     auto client = client_builder.build();
     if (!client) {
@@ -61,18 +71,34 @@
 
     client->async_connect();
 
-    entities_.emplace(id, std::make_pair(client, poster));
+    entities_.emplace(id, std::make_tuple(client, poster, suppress_next_error));
     return id;
 }
 
+std::optional<bool> EntityManager::command(std::string const& id,
+                                           std::string const& command) {
+    auto it = entities_.find(id);
+    if (it == entities_.end()) {
+        return std::nullopt;
+    }
+
+    if (command == "restart") {
+        std::get<2>(it->second)->store(true);
+        std::get<0>(it->second)->async_restart("test harness restart command");
+        return true;
+    }
+
+    return false;
+}
+
 bool EntityManager::destroy(std::string const& id) {
     auto it = entities_.find(id);
     if (it == entities_.end()) {
         return false;
     }
 
-    it->second.first->async_shutdown(nullptr);
-    it->second.second->stop();
+    std::get<0>(it->second)->async_shutdown(nullptr);
+    std::get<1>(it->second)->stop();
 
     entities_.erase(it);
 

contract-tests/sse-contract-tests/src/main.cpp

@@ -1,4 +1,4 @@
 #include "server.hpp"
 
 #include <launchdarkly/logging/console_backend.hpp>
 
@@ -17,7 +17,7 @@
 using launchdarkly::LogLevel;
 
 int main(int argc, char* argv[]) {
     launchdarkly::Logger logger{
         std::make_unique<ConsoleBackend>("sse-contract-tests")};
 
     std::string const default_port = "8123";
@@ -30,14 +30,15 @@
     try {
         net::io_context ioc{1};
 
         server srv(ioc, "0.0.0.0", boost::lexical_cast<unsigned short>(port),
                    logger);
 
         srv.add_capability("headers");
         srv.add_capability("comments");
         srv.add_capability("report");
         srv.add_capability("post");
-        srv.add_capability("reconnection");
+        srv.add_capability("restart");
+        srv.add_capability("server-directed-shutdown-request");
         srv.add_capability("read-timeout");
 
         net::signal_set signals{ioc, SIGINT, SIGTERM};

contract-tests/sse-contract-tests/src/session.cpp

@@ -1,4 +1,4 @@
 #include "session.hpp"
 #include <boost/algorithm/string/erase.hpp>
 #include <boost/asio/dispatch.hpp>
 #include <iostream>
@@ -101,6 +101,33 @@
         }
     }
 
+    if (req.method() == http::verb::post &&
+        req.target().starts_with(kEntityPath)) {
+        std::string entity_id = req.target();
+        boost::erase_first(entity_id, kEntityPath);
+
+        std::string command;
+        try {
+            auto json = nlohmann::json::parse(req.body());
+            command = json.at("command").get<std::string>();
+        } catch (nlohmann::json::exception& e) {
+            return bad_request("unable to parse command JSON");
+        }
+
+        auto handled = manager_.command(entity_id, command);
+        if (!handled) {
+            return not_found(req.target());
+        }
+        if (!*handled) {
+            return bad_request("unrecognized command");
+        }
+
+        Response res{http::status::ok, req.version()};
+        res.keep_alive(req.keep_alive());
+        res.prepare_payload();
+        return res;
+    }
+
     if (req.method() == http::verb::delete_ &&
         req.target().starts_with(kEntityPath)) {
         std::string entity_id = req.target();