diff --git a/config/das.json b/config/das.json index 0d316700..46fa7016 100644 --- a/config/das.json +++ b/config/das.json @@ -207,7 +207,8 @@ "thread_pool_size": 4, "max_concurrent_executions": 100, "max_queued_executions": 500, - "max_events_per_execution": 10000, + "max_events_per_execution": 100000, + "stream_items_per_chunk": 100, "execution_retention_ms": 900000 } } diff --git a/src/agents/command_router/BUILD b/src/agents/command_router/BUILD index 9d49a303..87858001 100644 --- a/src/agents/command_router/BUILD +++ b/src/agents/command_router/BUILD @@ -48,7 +48,6 @@ cc_library( deps = [ ":bus_command_router_proxy", ":evolution_metta_parser", - "//agents/command_router/http_api:command_router_http_api_singleton", "//agents/evolution:query_evolution_proxy", "//agents/query_engine:pattern_matching_query_proxy", "//agents/query_engine:query_answer", diff --git a/src/agents/command_router/BusCommandRouterProcessor.cc b/src/agents/command_router/BusCommandRouterProcessor.cc index 24f4521b..b3fe35d4 100644 --- a/src/agents/command_router/BusCommandRouterProcessor.cc +++ b/src/agents/command_router/BusCommandRouterProcessor.cc @@ -1,11 +1,13 @@ #include "BusCommandRouterProcessor.h" #include +#include #include #include "BaseQueryProxy.h" #include "EvolutionMettaParser.h" #include "PatternMatchingQueryProxy.h" +#include "PortPool.h" #include "QueryEvolutionProxy.h" #include "ServiceBusSingleton.h" #include "Utils.h" @@ -21,6 +23,7 @@ using namespace service_bus; namespace { const string CONTEXT_KEY = "context"; +atomic HTTP_REQUEST_SERIAL{1}; } // namespace @@ -44,6 +47,48 @@ Properties& BusCommandRouterProcessor::parameters_for_peer(const string& peer_id return iterator->second; } +void BusCommandRouterProcessor::dispatch_http_command( + const shared_ptr& caller_proxy, const string& http_requestor_id) { + if (caller_proxy == nullptr) { + RAISE_ERROR("HTTP command requires a caller proxy"); + } + if (caller_proxy->issued) { + RAISE_ERROR("Attempt to dispatch the same HTTP caller proxy twice"); + } + if (http_requestor_id.empty()) { + RAISE_ERROR("HTTP command dispatch requires a non-empty http_requestor_id"); + } + + const unsigned int serial = HTTP_REQUEST_SERIAL.fetch_add(1); + + caller_proxy->issued = true; + caller_proxy->requestor_id = http_requestor_id; + caller_proxy->serial = serial; + caller_proxy->proxy_port = PortPool::get_port(); + if (caller_proxy->proxy_port == 0) { + RAISE_ERROR("No port is available to start HTTP caller proxy"); + } + caller_proxy->setup_proxy_node(); + + auto processor_proxy = dynamic_pointer_cast(factory_empty_proxy()); + if (processor_proxy == nullptr) { + RAISE_ERROR("Invalid proxy type for HTTP BUS_COMMAND_ROUTER dispatch"); + } + processor_proxy->proxy_port = PortPool::get_port(); + if (processor_proxy->proxy_port == 0) { + RAISE_ERROR("No port is available to start HTTP processor proxy"); + } + processor_proxy->requestor_id = http_requestor_id; + processor_proxy->serial = serial; + const string requestor_host = http_requestor_id.substr(0, http_requestor_id.find(':')); + const string processor_proxy_node_id = requestor_host + ":" + to_string(processor_proxy->proxy_port); + processor_proxy->setup_proxy_node(processor_proxy_node_id, caller_proxy->my_id()); + processor_proxy->command = std::move(caller_proxy->command); + processor_proxy->args = std::move(caller_proxy->args); + + this->run_command(processor_proxy); +} + void BusCommandRouterProcessor::run_command(shared_ptr proxy) { auto router_proxy = dynamic_pointer_cast(proxy); if (router_proxy == nullptr) { diff --git a/src/agents/command_router/BusCommandRouterProcessor.h b/src/agents/command_router/BusCommandRouterProcessor.h index ab8219f6..f8b8b33f 100644 --- a/src/agents/command_router/BusCommandRouterProcessor.h +++ b/src/agents/command_router/BusCommandRouterProcessor.h @@ -31,6 +31,10 @@ class BusCommandRouterProcessor : public BusCommandProcessor { shared_ptr factory_empty_proxy() override; void run_command(shared_ptr proxy) override; + /** Run a router command in-process for HTTP; responses are written to caller_proxy. */ + void dispatch_http_command(const shared_ptr& caller_proxy, + const string& http_requestor_id); + private: void handle_get(shared_ptr proxy, const string& arg); void handle_set(shared_ptr proxy, const string& arg); diff --git a/src/agents/command_router/http_api/BUILD b/src/agents/command_router/http_api/BUILD index ae49685e..9758010d 100644 --- a/src/agents/command_router/http_api/BUILD +++ b/src/agents/command_router/http_api/BUILD @@ -12,6 +12,18 @@ cc_library( ], ) +cc_library( + name = "bus_command_router_proxy_stream_poller", + srcs = ["BusCommandRouterProxyStreamPoller.cc"], + hdrs = ["BusCommandRouterProxyStreamPoller.h"], + includes = ["."], + deps = [ + "//agents:base_query_proxy", + "//agents/command_router:bus_command_router_proxy", + "//commons:commons_lib", + ], +) + cc_library( name = "command_router_http_api_config", srcs = ["CommandRouterHttpAPIConfig.cc"], @@ -29,8 +41,11 @@ cc_library( hdrs = ["CommandRouterHttpAPI.h"], includes = ["."], deps = [ + ":bus_command_router_proxy_stream_poller", ":command_execution", ":command_router_http_api_config", + "//agents/command_router:bus_command_router_processor", + "//agents/command_router:bus_command_router_proxy", "//commons:commons_lib", "//commons/processor:processor_lib", "//hasher:hasher_lib", @@ -45,7 +60,8 @@ cc_library( deps = [ ":command_router_http_api", ":command_router_http_api_config", - "//commons:commons_lib", + "//agents/command_router:bus_command_router_processor", "//commons/processor:processor_lib", + "//service_bus:bus_command_processor", ], ) diff --git a/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc new file mode 100644 index 00000000..9c78ae7d --- /dev/null +++ b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.cc @@ -0,0 +1,171 @@ +#include "BusCommandRouterProxyStreamPoller.h" + +#include "BaseQueryProxy.h" +#include "Utils.h" + +using namespace command_router; +using namespace commons; +using namespace agents; + +namespace { + +void emit_chunk(const function& chunk)>& on_chunk, + vector& chunk_data) { + if (!chunk_data.empty() && on_chunk) { + on_chunk(chunk_data); + chunk_data.clear(); + } +} + +void append_answer_chunk(const shared_ptr& router_proxy, + bool use_metta_as_query_tokens, + size_t items_per_chunk, + const function& chunk)>& on_chunk, + vector& chunk_data) { + shared_ptr answer; + while ((answer = router_proxy->pop()) != nullptr) { + chunk_data.push_back(answer->to_string(use_metta_as_query_tokens)); + if (chunk_data.size() >= items_per_chunk) { + emit_chunk(on_chunk, chunk_data); + } + } + emit_chunk(on_chunk, chunk_data); +} + +} // namespace + +bool BusCommandRouterProxyStreamPoller::poll_stream( + const shared_ptr& router_proxy, + const string& command_type, + size_t items_per_chunk, + const function& should_abort, + const function& chunk)>& on_chunk, + const function& on_error, + const function& on_aborted) { + if (items_per_chunk == 0) { + if (on_error) { + on_error("items_per_chunk must be at least 1"); + } + return false; + } + + if (!router_proxy) { + if (on_error) { + on_error("router_proxy must not be null"); + } + return false; + } + + auto finished_or_error = [&]() { return router_proxy->finished() || router_proxy->error_flag; }; + + auto handle_abort = [&]() -> bool { + if (should_abort && should_abort()) { + router_proxy->abort(); + if (on_aborted) { + on_aborted(); + } + return true; + } + return false; + }; + + if (command_type == "get") { + while (router_proxy->params_response.empty() && !finished_or_error()) { + if (handle_abort()) { + return false; + } + Utils::sleep(100); + } + if (router_proxy->error_flag) { + if (on_error) { + on_error(router_proxy->error_message); + } + return false; + } + if (router_proxy->params_response.empty()) { + if (on_error) { + on_error("GET command finished without params response"); + } + return false; + } + if (on_chunk) { + on_chunk({router_proxy->params_response}); + } + return true; + } + + if (command_type == "set") { + while (router_proxy->set_param_ack.empty() && !finished_or_error()) { + if (handle_abort()) { + return false; + } + Utils::sleep(100); + } + if (router_proxy->error_flag) { + if (on_error) { + on_error(router_proxy->error_message); + } + return false; + } + if (router_proxy->set_param_ack.empty()) { + if (on_error) { + on_error("SET command finished without parameter ack"); + } + return false; + } + if (on_chunk) { + on_chunk({router_proxy->set_param_ack}); + } + return true; + } + + if (command_type == "query" || command_type == "evolution") { + while (!router_proxy->routed_flag && !finished_or_error()) { + if (handle_abort()) { + return false; + } + Utils::sleep(100); + } + if (router_proxy->error_flag) { + if (on_error) { + on_error(router_proxy->error_message); + } + return false; + } + if (!router_proxy->routed_flag) { + if (on_error) { + on_error("Command finished without being routed to a downstream service"); + } + return false; + } + + const bool use_metta_as_query_tokens = + router_proxy->parameters.get_or(BaseQueryProxy::USE_METTA_AS_QUERY_TOKENS, true); + + vector chunk_data; + while (!finished_or_error()) { + if (handle_abort()) { + return false; + } + + append_answer_chunk( + router_proxy, use_metta_as_query_tokens, items_per_chunk, on_chunk, chunk_data); + Utils::sleep(100); + } + append_answer_chunk( + router_proxy, use_metta_as_query_tokens, items_per_chunk, on_chunk, chunk_data); + + if (router_proxy->error_flag) { + if (on_error) { + on_error(router_proxy->error_message); + } + return false; + } + return true; + } + + if (on_error) { + on_error("Unknown command_type: " + command_type); + } + return false; +} diff --git a/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h new file mode 100644 index 00000000..526a8419 --- /dev/null +++ b/src/agents/command_router/http_api/BusCommandRouterProxyStreamPoller.h @@ -0,0 +1,53 @@ +#pragma once + +#include +#include +#include +#include + +#include "BusCommandRouterProxy.h" + +using namespace std; + +namespace command_router { + +/** + * Polls a BusCommandRouterProxy and delivers results in HTTP-friendly chunks. + * + * Used by CommandRouterHttpAPI to stream command output over WebSocket. The terminal + * bus client keeps its own inline polling loop instead of this helper. + */ +class BusCommandRouterProxyStreamPoller { + public: + /** + * Poll router_proxy until the command finishes, is aborted, or fails. + * + * For command_type "query" and "evolution", answers are popped from the proxy + * and forwarded in batches of at most items_per_chunk strings. For "get" and + * "set", a single chunk is emitted once the proxy response is ready. + * + * @param router_proxy Proxy already issued on the service bus. + * @param command_type Router command: "get", "set", "query", or "evolution". + * @param items_per_chunk Maximum answers per on_chunk call for query/evolution. + * Must be at least 1. + * @param should_abort Optional callback; when it returns true, the proxy is aborted + * and on_aborted is invoked. + * @param on_chunk Called with each batch of serialized answers or response payload. + * @param on_error Called with an error message on validation, proxy, or unknown + * command failures. + * @param on_aborted Called when polling stops because should_abort returned true. + * @return true when the command completed without error; false otherwise. + */ + static bool poll_stream(const shared_ptr& router_proxy, + const string& command_type, + size_t items_per_chunk, + const function& should_abort, + const function& chunk)>& on_chunk, + const function& on_error, + const function& on_aborted); + + private: + BusCommandRouterProxyStreamPoller() = delete; +}; + +} // namespace command_router diff --git a/src/agents/command_router/http_api/CommandExecution.cc b/src/agents/command_router/http_api/CommandExecution.cc index e5d762dc..77b4157d 100644 --- a/src/agents/command_router/http_api/CommandExecution.cc +++ b/src/agents/command_router/http_api/CommandExecution.cc @@ -57,7 +57,7 @@ int CommandExecution::received_count() const { return this->received_count_; } -long long CommandExecution::finished_at_ms() const { +unsigned long CommandExecution::finished_at_ms() const { lock_guard lock(this->mtx_); return this->finished_at_ms_; } @@ -111,7 +111,7 @@ optional CommandExecution::wait_next_event(size_t& next_index, return payload; } -bool CommandExecution::is_retention_expired(long long now_ms, long long retention_ms) const { +bool CommandExecution::is_retention_expired(unsigned long now_ms, unsigned long retention_ms) const { lock_guard lock(this->mtx_); return is_terminal(this->status_) && this->finished_at_ms_ > 0 && (now_ms - this->finished_at_ms_) > retention_ms; @@ -151,7 +151,7 @@ void CommandExecution::publish_chunk(int seq, const vector& data) { {"received_count", this->received_count_}}); } -void CommandExecution::mark_completed(long long duration_ms, int total_items) { +void CommandExecution::mark_completed(unsigned long duration_ms, int total_items) { lock_guard lock(this->mtx_); this->duration_ms_ = duration_ms; this->total_items_ = total_items; diff --git a/src/agents/command_router/http_api/CommandExecution.h b/src/agents/command_router/http_api/CommandExecution.h index fe1c8c1e..d878575c 100644 --- a/src/agents/command_router/http_api/CommandExecution.h +++ b/src/agents/command_router/http_api/CommandExecution.h @@ -53,7 +53,7 @@ class CommandExecution { bool is_terminal_status() const; bool is_cancel_requested() const; int received_count() const; - long long finished_at_ms() const; + unsigned long finished_at_ms() const; /** @brief JSON body for GET /executions/{id}. */ json to_status_json() const; @@ -73,7 +73,7 @@ class CommandExecution { bool& stream_finished); /** @brief True when terminal and finished_at_ms is older than retention_ms. */ - bool is_retention_expired(long long now_ms, long long retention_ms) const; + bool is_retention_expired(unsigned long now_ms, unsigned long retention_ms) const; /** @brief Append a chunk event and update received_count. */ void publish_chunk(int seq, const vector& data); @@ -82,7 +82,7 @@ class CommandExecution { void mark_running(); /** @brief -> COMPLETED; sets duration/totals and emits a lifecycle event. */ - void mark_completed(long long duration_ms, int total_items); + void mark_completed(unsigned long duration_ms, int total_items); /** @brief -> ERROR; sets error_message and emits a lifecycle event. */ void mark_error(const string& message); @@ -100,8 +100,8 @@ class CommandExecution { ExecutionStatus status_ = ExecutionStatus::PENDING; int received_count_ = 0; int total_items_ = 0; - long long duration_ms_ = 0; - long long finished_at_ms_ = 0; + unsigned long duration_ms_ = 0; + unsigned long finished_at_ms_ = 0; string error_message_; bool cancel_requested_ = false; vector events_; diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc index d6b09209..103ff585 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.cc @@ -4,6 +4,10 @@ #include #include +#include "BusCommandRouterProcessor.h" +#include "BusCommandRouterProxy.h" +#include "BusCommandRouterProxyStreamPoller.h" + #define LOG_LEVEL INFO_LEVEL #include "Logger.h" #include "Utils.h" @@ -12,6 +16,7 @@ using namespace command_router; using namespace commons; using namespace processor; +using namespace agents; using json = nlohmann::json; @@ -24,12 +29,16 @@ const unordered_set CommandRouterHttpAPI::VALID_COMMAND_TYPES = { CommandRouterHttpAPI::CommandRouterHttpAPI(const string& host, int port, shared_ptr thread_pool, - HttpAPISettings settings) + shared_ptr router_processor, + HttpAPISettings settings, + const string& bus_host) : Processor("command_router_http_api"), host(host), port(port), thread_pool(thread_pool), - settings(std::move(settings)) { + router_processor(router_processor), + settings(settings), + bus_host(bus_host) { if (this->thread_pool == nullptr) { RAISE_ERROR("CommandRouterHttpAPI requires a non-null thread pool"); } @@ -280,71 +289,95 @@ shared_ptr CommandRouterHttpAPI::find_execution(const string& } void CommandRouterHttpAPI::run_execution_inner(const shared_ptr& exec) { - // POC code to simulate event streaming. + if (this->router_processor == nullptr) { + exec->mark_error("Command router processor is not configured for HTTP execution"); + return; + } + + StopWatch timer; + timer.start(); - const int TOTAL = 1000; - const int CHUNK_SIZE = 10; - auto started_at = Utils::get_current_time_millis(); int seq = 0; - for (int i = 0; i < TOTAL; i += CHUNK_SIZE) { - if (exec->is_cancel_requested() || this->shutting_down.load()) { - exec->mark_aborted(); - LOG_INFO("CommandRouter HTTP API execution aborted id=" << exec->execution_id); + auto should_abort = [&]() { return exec->is_cancel_requested() || this->shutting_down.load(); }; + auto on_chunk = [&](const vector& chunk) { exec->publish_chunk(++seq, chunk); }; + auto on_error = [&](const string& message) { exec->mark_error(message); }; + auto on_aborted = [&]() { + exec->mark_aborted(); + LOG_INFO("CommandRouter HTTP API execution aborted id=" << exec->execution_id); + }; + auto on_complete = [&](unsigned long duration_ms, int total_items) { + exec->mark_completed(duration_ms, total_items); + LOG_INFO("CommandRouter HTTP API execution completed id=" + << exec->execution_id << " duration_ms=" << duration_ms << " items=" << total_items); + }; + + try { + string router_arg = exec->command_text; + Utils::replace_all(router_arg, "%", "$"); + auto router_proxy = make_shared(exec->command_type, router_arg); + + string http_requestor_id = this->bus_host + ":http-" + exec->execution_id; + this->router_processor->dispatch_http_command(router_proxy, http_requestor_id); + + if (!BusCommandRouterProxyStreamPoller::poll_stream(router_proxy, + exec->command_type, + this->settings.stream_items_per_chunk, + should_abort, + on_chunk, + on_error, + on_aborted)) { return; } - Utils::sleep(200); - - vector chunk_data; - for (int k = i; k < min(i + CHUNK_SIZE, TOTAL); ++k) { - chunk_data.push_back("result-" + std::to_string(k)); - } + timer.stop(); - exec->publish_chunk(++seq, chunk_data); + on_complete(timer.milliseconds(), exec->received_count()); + } catch (const exception& e) { + on_error(e.what()); } - - const auto duration_ms = Utils::get_current_time_millis() - started_at; - exec->mark_completed(duration_ms, exec->received_count()); - - LOG_INFO("CommandRouter HTTP API execution completed id=" << exec->execution_id - << " duration_ms=" << duration_ms - << " items=" << exec->received_count()); } void CommandRouterHttpAPI::run_execution(const shared_ptr& exec) { + bool acquired_running_slot = false; + { - unique_lock semaphore(this->executions_mtx); - this->execution_slots_cv.wait(semaphore, [&] { + unique_lock lock(this->executions_mtx); + + // Block until a concurrent slot is free or the server is shutting down. + this->execution_slots_cv.wait(lock, [&] { return this->shutting_down.load() || this->running_executions < this->settings.max_concurrent_executions; }); - if (this->shutting_down.load()) { - exec->mark_error_unless_terminal("Server is shutting down"); - this->pending_executions--; - return; + this->pending_executions--; + + if (!this->shutting_down.load()) { + if (!exec->is_cancel_requested()) { + this->running_executions++; + acquired_running_slot = true; + } } + } - exec->mark_running(); - this->pending_executions--; - this->running_executions++; + if (this->shutting_down.load()) { + exec->mark_error_unless_terminal("Server is shutting down"); + return; } - try { - this->run_execution_inner(exec); - } catch (const exception& e) { - LOG_ERROR("CommandRouter HTTP API execution failed id=" << exec->execution_id << ": " - << e.what()); - exec->mark_error_unless_terminal(string("Execution failed: ") + e.what()); - } catch (...) { - LOG_ERROR("CommandRouter HTTP API execution failed id=" << exec->execution_id - << ": unknown error"); - exec->mark_error_unless_terminal("Execution failed: unknown error"); + if (!acquired_running_slot) { + exec->mark_aborted_unless_terminal(); + return; } - lock_guard semaphore(this->executions_mtx); - this->running_executions--; + exec->mark_running(); + + this->run_execution_inner(exec); + + { + lock_guard lock(this->executions_mtx); + this->running_executions--; + } this->execution_slots_cv.notify_one(); } diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPI.h b/src/agents/command_router/http_api/CommandRouterHttpAPI.h index 5c6d5ad9..19fed3d9 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPI.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPI.h @@ -24,6 +24,8 @@ using json = nlohmann::json; namespace command_router { +class BusCommandRouterProcessor; + /** * @brief HTTP + WebSocket server for asynchronous command execution. * @@ -33,6 +35,9 @@ namespace command_router { * POST /command-router/executions/{id}/cancel — request cancel * WS /command-router/ws/{id} — stream JSON events * + * Command execution uses the registered BusCommandRouterProcessor via in-process proxy + * dispatch (dispatch_http_command). Each HTTP execution owns a caller proxy for streaming. + * * Runs on a DedicatedThread: thread_one_step() blocks in listen() until stop(). * Each accepted command is enqueued on thread_pool so the listener stays free. * Terminal executions stay in executions until execution_retention_ms expires. @@ -51,7 +56,9 @@ class CommandRouterHttpAPI : public processor::Processor, public processor::Thre CommandRouterHttpAPI(const string& host, int port, shared_ptr thread_pool, - HttpAPISettings settings = {}); + shared_ptr router_processor, + HttpAPISettings settings = {}, + const string& bus_host = "localhost"); /** @brief Calls stop(). */ ~CommandRouterHttpAPI() override; @@ -77,6 +84,8 @@ class CommandRouterHttpAPI : public processor::Processor, public processor::Thre int port; httplib::Server server; shared_ptr thread_pool; + shared_ptr router_processor; + string bus_host; HttpAPISettings settings; atomic shutting_down{false}; unordered_map> executions; diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc index 3dbcd858..d3cfec73 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.cc @@ -14,14 +14,14 @@ static pair parse_host_port(const string& endpoint) { RAISE_ERROR("Invalid " + config_path + " configuration: endpoint must be :"); } - int port; - try { - port = stoi(tokens[1]); - } catch (const exception&) { - RAISE_ERROR("Invalid " + config_path + " configuration: port must be an integer"); + string host = tokens[0]; + string port = tokens[1]; + + if (!Utils::is_number(port)) { + RAISE_ERROR("Invalid " + config_path + " configuration: port must be numeric"); } - return {tokens[0], port}; + return {host, stoi(port)}; } static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_config) { @@ -35,19 +35,29 @@ static HttpAPISettings load_http_api_settings(const JsonConfig& command_router_c command_router_config.at_path("http_api.max_events_per_execution") .get_or(settings.max_events_per_execution); settings.execution_retention_ms = command_router_config.at_path("http_api.execution_retention_ms") - .get_or(settings.execution_retention_ms); + .get_or(settings.execution_retention_ms); + settings.stream_items_per_chunk = command_router_config.at_path("http_api.stream_items_per_chunk") + .get_or(settings.stream_items_per_chunk); + if (settings.stream_items_per_chunk == 0) { + RAISE_ERROR( + "Invalid command_router.http_api.stream_items_per_chunk configuration: value must be at " + "least 1"); + } return settings; } CommandRouterHttpAPIConfig CommandRouterHttpAPIConfig::from_config( const JsonConfig& command_router_config) { - CommandRouterHttpAPIConfig config; - tie(config.host, config.port) = + auto [host, port] = parse_host_port(command_router_config.at_path("http_api.endpoint").get()); + auto [bus_host, _] = parse_host_port(command_router_config.at_path("endpoint").get()); + + CommandRouterHttpAPIConfig config; + config.host = host; + config.port = port; + config.bus_host = bus_host; config.settings = load_http_api_settings(command_router_config); config.thread_pool_size = command_router_config.at_path("http_api.thread_pool_size").get_or(4); - tie(config.bus_host, std::ignore) = - parse_host_port(command_router_config.at_path("endpoint").get()); return config; } diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h index d9e8f9b4..f4a8871a 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPIConfig.h @@ -14,7 +14,9 @@ struct HttpAPISettings { size_t max_concurrent_executions = 100; size_t max_queued_executions = 500; size_t max_events_per_execution = CommandExecution::DEFAULT_MAX_EVENTS; - long long execution_retention_ms = 15 * 60 * 1000; + unsigned long execution_retention_ms = 15 * 60 * 1000; + // Max query/evolution answers per WebSocket chunk event (1 = one item per event). + size_t stream_items_per_chunk = 1; }; struct CommandRouterHttpAPIConfig { diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc index 5e6e6b19..65751f33 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc +++ b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.cc @@ -1,5 +1,6 @@ #include "CommandRouterHttpAPISingleton.h" +#include "BusCommandRouterProcessor.h" #include "CommandRouterHttpAPIConfig.h" using namespace command_router; @@ -10,30 +11,43 @@ shared_ptr CommandRouterHttpAPISingleton::HTTP_API = shared_ptr(nullptr); mutex CommandRouterHttpAPISingleton::API_MUTEX; -void CommandRouterHttpAPISingleton::init(const JsonConfig& command_router_config) { +void CommandRouterHttpAPISingleton::init(const JsonConfig& command_router_config, + const shared_ptr& processor) { lock_guard semaphore(API_MUTEX); if (INITIALIZED) { RAISE_ERROR( "CommandRouterHttpAPISingleton already initialized. " "CommandRouterHttpAPISingleton::init() should be called only once."); } - CommandRouterHttpAPISingleton::create_and_start(command_router_config); + create_and_start(command_router_config, processor); } -void CommandRouterHttpAPISingleton::create_and_start(const JsonConfig& command_router_config) { +void CommandRouterHttpAPISingleton::create_and_start(const JsonConfig& command_router_config, + const shared_ptr& processor) { + auto router_processor = dynamic_pointer_cast(processor); + + if (router_processor == nullptr) { + RAISE_ERROR("CommandRouterHttpAPISingleton requires a non-null BusCommandRouterProcessor"); + } + const auto config = CommandRouterHttpAPIConfig::from_config(command_router_config); auto thread_pool_executor = make_shared("http_api_thread_pool_executor", config.thread_pool_size); - HTTP_API = make_shared( - config.host, config.port, thread_pool_executor, config.settings); + + HTTP_API = make_shared(config.host, + config.port, + thread_pool_executor, + router_processor, + config.settings, + config.bus_host); auto http_api_thread = make_shared("http_api_thread", HTTP_API.get()); try { CommandRouterHttpAPI::initialize(HTTP_API, {http_api_thread, thread_pool_executor}); INITIALIZED = true; - } catch (const std::exception& e) { + } catch (const exception& e) { CommandRouterHttpAPISingleton::shutdown_on_startup_failure(); RAISE_ERROR( "CommandRouterHttpAPISingleton::init(): CommandRouterHttpAPI::initialize() failed: " + diff --git a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h index 17ad5a1c..fc387146 100644 --- a/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h +++ b/src/agents/command_router/http_api/CommandRouterHttpAPISingleton.h @@ -3,16 +3,19 @@ #include #include +#include "BusCommandProcessor.h" #include "CommandRouterHttpAPI.h" #include "JsonConfig.h" using namespace std; +using namespace service_bus; namespace command_router { class CommandRouterHttpAPISingleton { public: - static void init(const JsonConfig& command_router_config); + static void init(const JsonConfig& command_router_config, + const shared_ptr& processor); static shared_ptr get_instance(); static void provide(shared_ptr http_api); @@ -21,7 +24,8 @@ class CommandRouterHttpAPISingleton { /** @brief Build and start HTTP API from command_router.http_api config. Sets HTTP_API and * INITIALIZED. */ - static void create_and_start(const JsonConfig& command_router_config); + static void create_and_start(const JsonConfig& command_router_config, + const shared_ptr& processor); /** @brief Stop and clear HTTP_API after a failed startup. */ static void shutdown_on_startup_failure(); diff --git a/src/main/BUILD b/src/main/BUILD index 43fd2c11..02894867 100644 --- a/src/main/BUILD +++ b/src/main/BUILD @@ -33,6 +33,7 @@ cc_library( deps = [ "//agents/atomdb_broker:atomdb_broker_lib", "//agents/command_router:command_router_lib", + "//agents/command_router/http_api:command_router_http_api_singleton", "//commons:commons_lib", "//main/helpers:main_helper_processor_factory_lib", ], diff --git a/src/main/bus_node.cc b/src/main/bus_node.cc index f278ff35..da865312 100644 --- a/src/main/bus_node.cc +++ b/src/main/bus_node.cc @@ -128,7 +128,7 @@ int main(int argc, char* argv[]) { mains::ProcessorType::COMMAND_ROUTER) { auto command_router_config = json_config.at_path("agents.command_router").get_or(JsonConfig()); - CommandRouterHttpAPISingleton::init(command_router_config); + CommandRouterHttpAPISingleton::init(command_router_config, service); } while (true) { diff --git a/src/service_bus/BusCommandProxy.h b/src/service_bus/BusCommandProxy.h index 2891ce89..cbdb6195 100644 --- a/src/service_bus/BusCommandProxy.h +++ b/src/service_bus/BusCommandProxy.h @@ -6,6 +6,10 @@ using namespace std; using namespace commons; using namespace distributed_algorithm_node; +namespace command_router { +class BusCommandRouterProcessor; +} // namespace command_router + namespace service_bus { class BusCommandProxy; @@ -83,6 +87,7 @@ class ProxyMessage : public Message { */ class BusCommandProxy { friend class ServiceBus; + friend class command_router::BusCommandRouterProcessor; public: // Commands allowed at the proxy level (caller <--> processor) diff --git a/src/tests/cpp/BUILD b/src/tests/cpp/BUILD index 619a1ea3..dd597164 100644 --- a/src/tests/cpp/BUILD +++ b/src/tests/cpp/BUILD @@ -1030,7 +1030,7 @@ cc_test( cc_test( name = "command_router_http_api_test", - size = "small", + size = "medium", srcs = [ "command_router_http_api_test.cc", ], @@ -1038,12 +1038,29 @@ cc_test( "-Iexternal/gtest/googletest/include", "-Iexternal/gtest/googletest", ], + linkopts = [ + "-L/usr/local/lib", + "-lpqxx", + "-lpq", + "-lhiredis_cluster", + "-lhiredis", + "-lmongocxx", + "-lbsoncxx", + ], linkstatic = 1, deps = [ + "//agents/command_router:bus_command_router_processor", + "//agents/command_router/http_api:bus_command_router_proxy_stream_poller", "//agents/command_router/http_api:command_execution", "//agents/command_router/http_api:command_router_http_api", + "//agents/command_router/http_api:command_router_http_api_config", "//agents/command_router/http_api:command_router_http_api_singleton", + "//agents/query_engine:query_answer", + "//atomdb:atomdb_singleton", "//commons/processor:processor_lib", + "//service_bus:service_bus_lib", + "//tests/cpp/test_commons:test_atomdb_json_config", + "//tests/cpp/test_commons:test_system_params", "@com_github_google_googletest//:gtest", "@mbedtls", ], diff --git a/src/tests/cpp/bus_command_router_test.cc b/src/tests/cpp/bus_command_router_test.cc index 19f35dd1..4c7acd76 100644 --- a/src/tests/cpp/bus_command_router_test.cc +++ b/src/tests/cpp/bus_command_router_test.cc @@ -1,7 +1,6 @@ #include "AtomDBSingleton.h" #include "BusCommandRouterProcessor.h" #include "BusCommandRouterProxy.h" -#include "CommandRouterHttpAPISingleton.h" #include "EvolutionMettaParser.h" #include "ServiceBus.h" #include "TestAtomDBJsonConfig.h" diff --git a/src/tests/cpp/command_router_http_api_test.cc b/src/tests/cpp/command_router_http_api_test.cc index 2e7dbfc1..64a724dc 100644 --- a/src/tests/cpp/command_router_http_api_test.cc +++ b/src/tests/cpp/command_router_http_api_test.cc @@ -1,11 +1,22 @@ -#include #include +#include "AtomDBSingleton.h" +#include "BaseProxy.h" +#include "BaseQueryProxy.h" +#include "BusCommandRouterProcessor.h" +#include "BusCommandRouterProxy.h" +#include "BusCommandRouterProxyStreamPoller.h" #include "CommandExecution.h" #include "CommandRouterHttpAPI.h" +#include "CommandRouterHttpAPIConfig.h" #include "CommandRouterHttpAPISingleton.h" #include "DedicatedThread.h" #include "JsonConfig.h" +#include "PortPool.h" +#include "QueryAnswer.h" +#include "ServiceBus.h" +#include "TestAtomDBJsonConfig.h" +#include "TestSystemParams.h" #include "gtest/gtest.h" #include "httplib.h" #include "processor/ThreadPool.h" @@ -13,6 +24,10 @@ using namespace command_router; using namespace processor; using namespace commons; +using namespace atomdb; +using namespace query_engine; +using namespace service_bus; +using das_test::init_test_system_parameters_singleton; using json = nlohmann::json; namespace { @@ -28,11 +43,57 @@ json make_execution_body(const string& command_type = "query", return {{"command_type", command_type}, {"command_text", command_text}}; } +class HangingQueryForwardProxy : public BusCommandProxy { + public: + void pack_command_line_args() override {} +}; + +/** Accepts forwarded queries but never responds, so HTTP executions stay running. */ +class HangingQueryForwardProcessor : public BusCommandProcessor { + public: + HangingQueryForwardProcessor() : BusCommandProcessor({ServiceBus::PATTERN_MATCHING_QUERY}) {} + + shared_ptr factory_empty_proxy() override { + return make_shared(); + } + + void run_command(shared_ptr /*proxy*/) override {} +}; + +void initialize_test_service_bus_statics_once() { + static bool initialized = false; + if (!initialized) { + ServiceBus::initialize_statics( + {ServiceBus::BUS_COMMAND_ROUTER, ServiceBus::PATTERN_MATCHING_QUERY}, 49400, 49999); + initialized = true; + } +} + +/** + * HTTP API fixture with an in-process command router bus so executions reach "running". + */ class HttpAPIServerFixture { public: void start(int port, const HttpAPISettings& settings = {}, unsigned int num_threads = 8) { + initialize_test_service_bus_statics_once(); + + const unsigned int query_port = PortPool::get_port(); + const string query_id = TEST_HOST + ":" + std::to_string(query_port); + const unsigned int router_port = PortPool::get_port(); + const string router_id = TEST_HOST + ":" + std::to_string(router_port); + + this->query_bus = make_shared(query_id); + this->query_bus->register_processor(make_shared()); + Utils::sleep(300); + + this->router_bus = make_shared(router_id, query_id); + this->router_processor = make_shared(this->router_bus); + this->router_bus->register_processor(this->router_processor); + Utils::sleep(500); + this->thread_pool = make_shared("test_thread_pool", num_threads); - this->api = make_shared(TEST_HOST, port, this->thread_pool, settings); + this->api = make_shared( + TEST_HOST, port, this->thread_pool, this->router_processor, settings, TEST_HOST); this->api_thread = make_shared("test_api_thread", this->api.get()); CommandRouterHttpAPI::initialize(this->api, {this->api_thread, this->thread_pool}); Utils::sleep(300); @@ -42,6 +103,12 @@ class HttpAPIServerFixture { if (this->api != nullptr) { this->api->stop(); } + this->api_thread = nullptr; + this->thread_pool = nullptr; + this->api = nullptr; + this->router_processor = nullptr; + this->router_bus = nullptr; + this->query_bus = nullptr; } httplib::Client make_client(int port) const { @@ -52,11 +119,71 @@ class HttpAPIServerFixture { } private: + shared_ptr query_bus; + shared_ptr router_bus; + shared_ptr router_processor; shared_ptr thread_pool; shared_ptr api; shared_ptr api_thread; }; +class CommandRouterStreamTestEnvironment : public ::testing::Environment { + public: + void SetUp() override { + AtomDBSingleton::init(test_atomdb_json_config()); + init_test_system_parameters_singleton(); + } +}; + +class StreamTestProxy : public BusCommandRouterProxy { + public: + StreamTestProxy() : BusCommandRouterProxy("query", "(Similarity %V1 %V2)") {} + + void enqueue_answers(unsigned int count) { + vector bundle; + for (unsigned int i = 0; i < count; ++i) { + QueryAnswer answer("answer-" + std::to_string(i), 0.0); + bundle.push_back(answer.tokenize()); + } + this->from_remote_peer(BaseQueryProxy::ANSWER_BUNDLE, bundle); + } + + void mark_routed() { this->from_remote_peer(BusCommandRouterProxy::ROUTED, {}); } + + void mark_finished() { this->from_remote_peer(BaseProxy::FINISHED, {}); } +}; + +size_t total_items(const vector>& chunks) { + size_t count = 0; + for (const auto& chunk : chunks) { + count += chunk.size(); + } + return count; +} + +vector chunk_sizes(const vector>& chunks) { + vector sizes; + for (const auto& chunk : chunks) { + sizes.push_back(chunk.size()); + } + return sizes; +} + +JsonConfig make_command_router_config(const json& overrides = json::object()) { + json root = {{"endpoint", "localhost:40008"}, + {"ports_range", "48000:48999"}, + {"http_api", + {{"endpoint", "localhost:40009"}, + {"thread_pool_size", 8}, + {"max_concurrent_executions", 50}, + {"max_queued_executions", 200}, + {"max_events_per_execution", 5000}, + {"stream_items_per_chunk", 25}, + {"execution_retention_ms", 123456}}}}; + root.merge_patch(overrides); + return JsonConfig(root); +} + } // namespace class CommandRouterHttpAPITest : public ::testing::Test { @@ -134,6 +261,169 @@ TEST(CommandExecutionTest, event_buffer_overflow_raises) { EXPECT_THROW(exec.publish_chunk(2, {"b", "c"}), runtime_error); } +// ----------------------------------------------------------------------------- +// CommandRouterHttpAPIConfig + +TEST(CommandRouterHttpAPIConfigTest, from_config_loads_http_api_fields) { + const auto config = CommandRouterHttpAPIConfig::from_config(make_command_router_config()); + + EXPECT_EQ(config.host, "localhost"); + EXPECT_EQ(config.port, 40009); + EXPECT_EQ(config.thread_pool_size, 8u); + EXPECT_EQ(config.bus_host, "localhost"); + EXPECT_EQ(config.settings.max_concurrent_executions, 50u); + EXPECT_EQ(config.settings.max_queued_executions, 200u); + EXPECT_EQ(config.settings.max_events_per_execution, 5000u); + EXPECT_EQ(config.settings.execution_retention_ms, 123456); + EXPECT_EQ(config.settings.stream_items_per_chunk, 25u); +} + +TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_zero_stream_items_per_chunk) { + EXPECT_THROW(CommandRouterHttpAPIConfig::from_config( + make_command_router_config({{"http_api", {{"stream_items_per_chunk", 0}}}})), + runtime_error); +} + +TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_invalid_http_api_endpoint) { + EXPECT_THROW(CommandRouterHttpAPIConfig::from_config( + make_command_router_config({{"http_api", {{"endpoint", "not-a-valid-endpoint"}}}})), + runtime_error); +} + +TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_non_numeric_http_api_port) { + EXPECT_THROW(CommandRouterHttpAPIConfig::from_config( + make_command_router_config({{"http_api", {{"endpoint", "localhost:abc"}}}})), + runtime_error); +} + +TEST(CommandRouterHttpAPIConfigTest, from_config_rejects_trailing_junk_http_api_port) { + EXPECT_THROW(CommandRouterHttpAPIConfig::from_config( + make_command_router_config({{"http_api", {{"endpoint", "localhost:8080abc"}}}})), + runtime_error); +} + +// ----------------------------------------------------------------------------- +// BusCommandRouterProcessor HTTP dispatch + +TEST(BusCommandRouterProcessorTest, dispatch_http_command_get_returns_params) { + set commands = {ServiceBus::BUS_COMMAND_ROUTER}; + ServiceBus::initialize_statics(commands, 49400, 49499); + initialize_test_service_bus_statics_once(); + + const string router_id = TEST_HOST + ":" + std::to_string(PortPool::get_port()); + auto router_bus = make_shared(router_id); + auto router_processor = make_shared(router_bus); + router_bus->register_processor(router_processor); + Utils::sleep(500); + + auto caller_proxy = make_shared("get", "params"); + router_processor->dispatch_http_command(caller_proxy, TEST_HOST + ":http-get-test"); + Utils::sleep(500); + + EXPECT_FALSE(caller_proxy->params_response.empty()); + EXPECT_TRUE(caller_proxy->finished()); +} + +// ----------------------------------------------------------------------------- +// BusCommandRouterProxyStreamPoller (HTTP stream polling) + +TEST(BusCommandRouterProxyStreamPollerTest, stream_emits_one_item_per_chunk_by_default) { + auto proxy = make_shared(); + proxy->mark_routed(); + proxy->enqueue_answers(5); + proxy->mark_finished(); + + vector> chunks; + auto on_chunk = [&](const vector& chunk) { chunks.push_back(chunk); }; + + ASSERT_TRUE(BusCommandRouterProxyStreamPoller::poll_stream( + proxy, "query", 1, nullptr, on_chunk, nullptr, nullptr)); + EXPECT_EQ(chunks.size(), 5u); + EXPECT_EQ(chunk_sizes(chunks), (vector{1, 1, 1, 1, 1})); + EXPECT_EQ(total_items(chunks), 5u); + EXPECT_NE(chunks.front().front().find("answer-0"), string::npos); +} + +TEST(BusCommandRouterProxyStreamPollerTest, stream_groups_items_per_chunk) { + auto proxy = make_shared(); + proxy->mark_routed(); + proxy->enqueue_answers(7); + proxy->mark_finished(); + + vector> chunks; + auto on_chunk = [&](const vector& chunk) { chunks.push_back(chunk); }; + + ASSERT_TRUE(BusCommandRouterProxyStreamPoller::poll_stream( + proxy, "query", 3, nullptr, on_chunk, nullptr, nullptr)); + EXPECT_EQ(chunks.size(), 3u); + EXPECT_EQ(chunk_sizes(chunks), (vector{3, 3, 1})); + EXPECT_EQ(total_items(chunks), 7u); +} + +TEST(BusCommandRouterProxyStreamPollerTest, rejects_zero_items_per_chunk) { + auto proxy = make_shared(); + proxy->mark_routed(); + proxy->enqueue_answers(1); + proxy->mark_finished(); + + string error_message; + auto on_error = [&](const string& message) { error_message = message; }; + + EXPECT_FALSE(BusCommandRouterProxyStreamPoller::poll_stream( + proxy, "query", 0, nullptr, nullptr, on_error, nullptr)); + EXPECT_NE(error_message.find("items_per_chunk"), string::npos); +} + +TEST(BusCommandRouterProxyStreamPollerTest, get_and_set_emit_single_chunk) { + auto get_proxy = make_shared(); + get_proxy->from_remote_peer(BusCommandRouterProxy::PARAMS_RESPONSE, {"params-body"}); + + vector> get_chunks; + ASSERT_TRUE(BusCommandRouterProxyStreamPoller::poll_stream( + get_proxy, + "get", + 1, + nullptr, + [&](const vector& chunk) { get_chunks.push_back(chunk); }, + nullptr, + nullptr)); + ASSERT_EQ(get_chunks.size(), 1u); + EXPECT_EQ(get_chunks[0], vector{"params-body"}); + + auto set_proxy = make_shared(); + set_proxy->from_remote_peer(BusCommandRouterProxy::SET_PARAM_ACK, {"ack-body"}); + + vector> set_chunks; + ASSERT_TRUE(BusCommandRouterProxyStreamPoller::poll_stream( + set_proxy, + "set", + 1, + nullptr, + [&](const vector& chunk) { set_chunks.push_back(chunk); }, + nullptr, + nullptr)); + ASSERT_EQ(set_chunks.size(), 1u); + EXPECT_EQ(set_chunks[0], vector{"ack-body"}); +} + +TEST(BusCommandRouterProxyStreamPollerTest, get_and_set_reject_empty_response) { + auto get_proxy = make_shared(); + get_proxy->mark_finished(); + + string get_error; + EXPECT_FALSE(BusCommandRouterProxyStreamPoller::poll_stream( + get_proxy, "get", 1, nullptr, nullptr, [&](const string& msg) { get_error = msg; }, nullptr)); + EXPECT_NE(get_error.find("params response"), string::npos); + + auto set_proxy = make_shared(); + set_proxy->mark_finished(); + + string set_error; + EXPECT_FALSE(BusCommandRouterProxyStreamPoller::poll_stream( + set_proxy, "set", 1, nullptr, nullptr, [&](const string& msg) { set_error = msg; }, nullptr)); + EXPECT_NE(set_error.find("parameter ack"), string::npos); +} + // ----------------------------------------------------------------------------- // HTTP API @@ -351,20 +641,24 @@ TEST_F(CommandRouterHttpAPITest, websocket_streams_lifecycle_events) { TEST_F(CommandRouterHttpAPISingletonTest, provide_and_get_instance) { auto pool = make_shared("test_pool", 1); - auto api = make_shared("localhost", 19002, pool); + auto api = make_shared("localhost", 19002, pool, nullptr); CommandRouterHttpAPISingleton::provide(api); EXPECT_EQ(CommandRouterHttpAPISingleton::get_instance().get(), api.get()); } TEST_F(CommandRouterHttpAPISingletonTest, init_after_provide_throws) { auto pool = make_shared("double_init_pool", 1); - CommandRouterHttpAPISingleton::provide(make_shared("localhost", 19005, pool)); + CommandRouterHttpAPISingleton::provide( + make_shared("localhost", 19005, pool, nullptr)); json raw = {{"http_api", {{"endpoint", "localhost:19005"}}}}; - EXPECT_THROW(CommandRouterHttpAPISingleton::init(JsonConfig(raw)), runtime_error); + EXPECT_THROW( + CommandRouterHttpAPISingleton::init(JsonConfig(raw), make_shared()), + runtime_error); } int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); + ::testing::AddGlobalTestEnvironment(new CommandRouterStreamTestEnvironment()); return RUN_ALL_TESTS(); }