fix(core): avoid socket timeouts

Author: maksimkurb

src/cmd/test_routing.cpp

@@ -450,39 +450,29 @@ std::optional<bool> test_rule_ipset_membership(const KernelSetTester& set_tester
     return false;
 }
 
-std::string find_actual_outbound(const KernelSetTester& set_tester,
-                                   const std::vector<RuleState>& rule_states,
-                                   const std::string& ip,
-                                   bool is_v4) {
+std::string find_actual_outbound(
+    const std::vector<RuleState>& rule_states,
+    const std::vector<RuleIpDiagnostic>& rule_ip_diagnostics) {
     bool any_answer = false;
 
-    for (const auto& rs : rule_states) {
+    const size_t count = std::min(rule_states.size(), rule_ip_diagnostics.size());
+    for (size_t idx = 0; idx < count; ++idx) {
+        const auto& rs = rule_states[idx];
         if (rs.action_type == RuleActionType::Skip) continue;
 
-        for (const auto& set_name : rs.set_names) {
-            bool v4_set = has_prefix(set_name, "kpbr4_") || has_prefix(set_name, "kpbr4s_") ||
-                          has_prefix(set_name, "kpbr4S_") || has_prefix(set_name, "kpbr4d_");
-            bool v6_set = has_prefix(set_name, "kpbr6_") || has_prefix(set_name, "kpbr6s_") ||
-                          has_prefix(set_name, "kpbr6S_") || has_prefix(set_name, "kpbr6d_");
-
-            if (is_v4 && !v4_set) continue;
-            if (!is_v4 && !v6_set) continue;
-
-            auto result = set_tester.contains(set_name, ip);
-            if (result.has_value()) {
-                any_answer = true;
-                if (*result) return rs.outbound_tag;
-            }
+        const auto& membership = rule_ip_diagnostics[idx].in_ipset;
+        if (membership.has_value()) {
+            any_answer = true;
+            if (*membership) return rs.outbound_tag;
         }
     }
 
     return any_answer ? "(default)" : "(unknown)";
 }
 
-// Kernel membership checks invoke nft/ipset subprocesses.  Keep a modest bound
-// so a domain with a very large DNS response cannot overwhelm the router, while
-// still checking different IPs concurrently.
-constexpr std::size_t kTestRoutingMaxConcurrentIps = 8;
+// Kernel membership checks invoke nft/ipset subprocesses. Keep the per-request
+// bound low because the daemon may run multiple routing tests concurrently.
+constexpr std::size_t kTestRoutingMaxConcurrentIps = 2;
 
 struct PerIpRoutingResult {
     TestRoutingEntry entry;
@@ -566,10 +556,6 @@ TestRoutingResult compute_test_routing(const Config& config,
         auto [expected, match] = find_expected_outbound(config, lookups, ip, domain_cands);
         per_ip.entry.expected_outbound = expected;
         per_ip.entry.list_match = std::move(match);
-        per_ip.entry.actual_outbound = set_tester.has_value()
-            ? find_actual_outbound(*set_tester, rule_states, ip, is_ipv4_address(ip))
-            : "(unknown)";
-        per_ip.entry.ok = (per_ip.entry.expected_outbound == per_ip.entry.actual_outbound);
 
         per_ip.rule_ip_diagnostics.reserve(result.rule_diagnostics.size());
         for (size_t idx = 0; idx < result.rule_diagnostics.size(); ++idx) {
@@ -581,6 +567,11 @@ TestRoutingResult compute_test_routing(const Config& config,
             }
             per_ip.rule_ip_diagnostics.push_back(std::move(ip_diag));
         }
+
+        per_ip.entry.actual_outbound = set_tester.has_value()
+            ? find_actual_outbound(rule_states, per_ip.rule_ip_diagnostics)
+            : "(unknown)";
+        per_ip.entry.ok = (per_ip.entry.expected_outbound == per_ip.entry.actual_outbound);
     };
 
     const size_t worker_count = std::min(kTestRoutingMaxConcurrentIps, ips.size());

src/daemon/daemon.hpp

@@ -160,6 +160,8 @@ class Daemon {
   void handle_control_commands();
   void setup_ipc_control_socket();
   void handle_ipc_control_socket();
+  bool try_begin_routing_test();
+  void finish_routing_test();
   void remove_ipc_control_socket() noexcept;
   void wake_control_loop();
   bool is_event_loop_thread() const;
@@ -380,12 +382,16 @@ IcmpTester icmp_tester_;
   BlockingExecutor resolver_stream_executor_{1, 16};
   BlockingExecutor resolver_io_executor_{1, 32};
   BlockingExecutor lifecycle_executor_{1, 16};
+  // Routing diagnostics are CPU/process-heavy. Two workers allow API and CLI
+  // tests to overlap while the small queue keeps resource use bounded.
+  BlockingExecutor routing_test_executor_{2, 2};
   std::atomic<std::uint64_t> runtime_generation_{1};
   std::atomic<bool> remote_list_refresh_inflight_{false};
   std::atomic<bool> ipc_mutation_inflight_{false};
   std::atomic<bool> ipc_resolver_hook_inflight_{false};
   std::atomic<bool> resolver_hash_refresh_inflight_{false};
   std::atomic<std::uint64_t> resolver_stream_completed_{0};
+  std::atomic<std::size_t> routing_tests_inflight_{0};
   TracedMutex system_resolver_hook_mutex_;
 
 #ifdef WITH_API

src/daemon/daemon_api.cpp

@@ -738,11 +738,31 @@ void Daemon::setup_api() {
         [this](const std::string& target) {
             const Config active_config = config_store_.active_config();
             const auto runtime_snapshot = runtime_state_store_.snapshot();
-            return compute_test_routing(
-                active_config,
-                list_service_.cache_manager(),
-                target,
-                &runtime_snapshot.firewall_state.get_rules());
+            const auto realized_rules = runtime_snapshot.firewall_state.get_rules();
+            if (!try_begin_routing_test()) {
+                throw ApiError("Too many routing tests are already running", 503);
+            }
+            auto pending = routing_test_executor_.try_submit(
+                "api-test-routing",
+                [this, active_config, realized_rules, target] {
+                    try {
+                        auto result = compute_test_routing(
+                            active_config,
+                            list_service_.cache_manager(),
+                            target,
+                            &realized_rules);
+                        finish_routing_test();
+                        return result;
+                    } catch (...) {
+                        finish_routing_test();
+                        throw;
+                    }
+                });
+            if (!pending.has_value()) {
+                finish_routing_test();
+                throw ApiError("Routing test executor queue is full", 503);
+            }
+            return pending->get();
         },
         [this]() {
             begin_config_operation_or_throw(ConfigOperationState::Saving,

src/daemon/daemon_core.cpp

@@ -56,6 +56,7 @@ constexpr auto INTERFACE_MONITOR_RECONNECT_RETRY_DELAY =
     std::chrono::seconds{5};
 constexpr std::size_t kResolverStreamChunkBytes =
     static_cast<std::size_t>(16) * 1024U;
+constexpr std::size_t kMaxConcurrentRoutingTests = 2;
 
 void send_all(int fd, const char *data, std::size_t size) {
   std::size_t written = 0;
@@ -220,6 +221,7 @@ Daemon::~Daemon() {
     resolver_hook_executor_.shutdown();
     resolver_stream_executor_.shutdown();
     resolver_io_executor_.shutdown();
+    routing_test_executor_.shutdown();
     blocking_executor_.shutdown();
 
     if (control_fd_ >= 0) {
@@ -324,6 +326,22 @@ void Daemon::remove_ipc_control_socket() noexcept {
   }
 }
 
+bool Daemon::try_begin_routing_test() {
+  std::size_t current = routing_tests_inflight_.load(std::memory_order_acquire);
+  while (current < kMaxConcurrentRoutingTests) {
+    if (routing_tests_inflight_.compare_exchange_weak(
+            current, current + 1, std::memory_order_acq_rel,
+            std::memory_order_acquire)) {
+      return true;
+    }
+  }
+  return false;
+}
+
+void Daemon::finish_routing_test() {
+  routing_tests_inflight_.fetch_sub(1, std::memory_order_acq_rel);
+}
+
 void Daemon::handle_ipc_control_socket() {
   while (true) {
     const int client = accept4(ipc_control_fd_, nullptr, nullptr, SOCK_CLOEXEC);
@@ -340,7 +358,7 @@ void Daemon::handle_ipc_control_socket() {
                      sizeof(timeout));
     nlohmann::json request = nlohmann::json::object();
     nlohmann::json response;
-    bool resolver_stream_dispatched = false;
+    bool client_dispatched = false;
     try {
       ucred peer{};
       socklen_t peer_length = sizeof(peer);
@@ -400,34 +418,78 @@ void Daemon::handle_ipc_control_socket() {
           const std::string target = request.value("target", "");
           if (target.empty())
             throw ipc::ControlProtocolError("test-routing requires a target");
+          const Config active_config = config_store_.active_config();
           const auto runtime_snapshot = runtime_state_store_.snapshot();
-          const auto result =
-              compute_test_routing(config_store_.active_config(),
-                                   list_service_.cache_manager(), target,
-                                   &runtime_snapshot.firewall_state.get_rules());
-          nlohmann::json entries = nlohmann::json::array();
-          for (const auto &entry : result.entries) {
-            nlohmann::json entry_json = {
-                {"ip", entry.ip},
-                {"expected_outbound", entry.expected_outbound},
-                {"actual_outbound", entry.actual_outbound},
-                {"ok", entry.ok}};
-            if (entry.list_match.has_value()) {
-              entry_json["list_match"] = {
-                  {"list_name", entry.list_match->list_name},
-                  {"via", entry.list_match->via}};
-            }
-            entries.push_back(std::move(entry_json));
+          const auto realized_rules = runtime_snapshot.firewall_state.get_rules();
+          const auto request_snapshot = request;
+          if (!try_begin_routing_test()) {
+            response = ipc::make_error_response(
+                request, "busy", "too many routing tests are already running");
+            const std::string frame = ipc::encode_message(response);
+            (void)send(client, frame.data(), frame.size(), MSG_NOSIGNAL);
+            close(client);
+            continue;
           }
-          response = {{"protocol_version", ipc::kControlProtocolVersion},
-                      {"request_id", request.at("request_id")},
+          const bool queued = routing_test_executor_.try_post(
+              "ipc-test-routing",
+              [this, client, active_config, realized_rules, request_snapshot,
+               target] {
+                nlohmann::json worker_response;
+                try {
+                  const auto result = compute_test_routing(
+                      active_config, list_service_.cache_manager(), target,
+                      &realized_rules);
+                  nlohmann::json entries = nlohmann::json::array();
+                  for (const auto &entry : result.entries) {
+                    nlohmann::json entry_json = {
+                        {"ip", entry.ip},
+                        {"expected_outbound", entry.expected_outbound},
+                        {"actual_outbound", entry.actual_outbound},
+                        {"ok", entry.ok}};
+                    if (entry.list_match.has_value()) {
+                      entry_json["list_match"] = {
+                          {"list_name", entry.list_match->list_name},
+                          {"via", entry.list_match->via}};
+                    }
+                    entries.push_back(std::move(entry_json));
+                  }
+                  worker_response = {
+                      {"protocol_version", ipc::kControlProtocolVersion},
+                      {"request_id", request_snapshot.at("request_id")},
                       {"ok", !result.dns_error.has_value()},
                       {"result",
                        {{"target", result.target},
                         {"resolved_ips", result.resolved_ips},
                         {"entries", std::move(entries)},
                         {"warnings", result.warnings},
                         {"dns_error", result.dns_error}}}};
+                } catch (const std::exception &error) {
+                  worker_response = ipc::make_error_response(
+                      request_snapshot, "daemon_error", error.what());
+                } catch (...) {
+                  worker_response = ipc::make_error_response(
+                      request_snapshot, "daemon_error",
+                      "routing test failed with an unknown error");
+                }
+
+                finish_routing_test();
+
+                try {
+                  const std::string frame = ipc::encode_message(worker_response);
+                  send_all(client, frame.data(), frame.size());
+                } catch (const std::exception &error) {
+                  Logger::instance().warn(
+                      "test-routing control response failed: {}", error.what());
+                }
+                close(client);
+              });
+          if (queued) {
+            client_dispatched = true;
+            continue;
+          }
+          finish_routing_test();
+          response = ipc::make_error_response(
+              request, "busy", "routing test executor queue is full");
         } else if (operation == "generate-resolver-config") {
           const RuntimeState runtime_state = runtime_state_machine_.state();
           // The DNS configuration is a daemon-owned desired-state
@@ -527,7 +589,7 @@ void Daemon::handle_ipc_control_socket() {
                 }
               });
           if (queued) {
-            resolver_stream_dispatched = true;
+            client_dispatched = true;
             continue;
           }
           response = ipc::make_error_response(
@@ -629,7 +691,7 @@ void Daemon::handle_ipc_control_socket() {
       response =
           ipc::make_error_response(request, "protocol_error", error.what());
     }
-    if (!resolver_stream_dispatched) {
+    if (!client_dispatched) {
       const std::string frame = ipc::encode_message(response);
       (void)send(client, frame.data(), frame.size(), MSG_NOSIGNAL);
       close(client);
@@ -1376,6 +1438,7 @@ void Daemon::run() {
   resolver_hook_executor_.shutdown();
   resolver_stream_executor_.shutdown();
   resolver_io_executor_.shutdown();
+  routing_test_executor_.shutdown();
   blocking_executor_.shutdown();
 
 #ifdef WITH_API

src/util/blocking_executor.hpp

@@ -8,6 +8,7 @@
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <pthread.h>
 #include <queue>
 #include <stdexcept>
@@ -31,6 +32,37 @@ class BlockingExecutor {
                   std::function<void()> task,
                   TraceId trace_id = current_trace_id());
 
+    template<typename Fn>
+    auto try_submit(std::string label,
+                    Fn&& fn,
+                    TraceId trace_id = current_trace_id())
+        -> std::optional<std::future<typename std::invoke_result_t<Fn>>> {
+        using Result = typename std::invoke_result_t<Fn>;
+
+        auto promise = std::make_shared<std::promise<Result>>();
+        auto future = promise->get_future();
+        const bool enqueued = enqueue(
+            std::move(label),
+            [promise, fn = std::forward<Fn>(fn)]() mutable {
+                try {
+                    if constexpr (std::is_void_v<Result>) {
+                        fn();
+                        promise->set_value();
+                    } else {
+                        promise->set_value(fn());
+                    }
+                } catch (...) {
+                    promise->set_exception(std::current_exception());
+                }
+            },
+            trace_id,
+            /*block_until_room=*/false);
+        if (!enqueued) {
+            return std::nullopt;
+        }
+        return std::optional<std::future<Result>>(std::move(future));
+    }
+
     template<typename Fn>
     auto submit(std::string label,
                 Fn&& fn,

tests/test_test_routing.cpp

@@ -353,12 +353,14 @@ TEST_CASE("compute_test_routing includes route rule conditions in diagnostics")
 TEST_CASE("compute_test_routing uses realized iptables generation set names") {
     const auto temp_dir = make_temp_dir();
     const auto bin_dir = temp_dir / "bin";
+    const auto invocation_log = temp_dir / "ipset-invocations.txt";
     std::filesystem::create_directories(bin_dir);
 
     write_executable(bin_dir / "iptables", "#!/bin/sh\nexit 0\n");
     write_executable(
         bin_dir / "ipset",
         "#!/bin/sh\n"
+        "echo test >> " + invocation_log.string() + "\n"
         "if [ \"$1\" = test ] && [ \"$2\" = kpbr4S_remote ] && "
         "[ \"$3\" = 203.0.113.10 ]; then\n"
         "  exit 0\n"
@@ -417,6 +419,12 @@ TEST_CASE("compute_test_routing uses realized iptables generation set names") {
     REQUIRE(result.rule_diagnostics.front().ip_rows.front().in_ipset.has_value());
     CHECK(*result.rule_diagnostics.front().ip_rows.front().in_ipset);
 
+    std::ifstream invocations(invocation_log);
+    const std::string invocation_contents{
+        std::istreambuf_iterator<char>(invocations),
+        std::istreambuf_iterator<char>()};
+    CHECK(invocation_contents == "test\n");
+
     std::filesystem::remove_all(temp_dir);
 }