add llarp.endpoint rpc command with kill option to kill all active convotags given

a list of or a single remote service address.

llarp/rpc/rpc_server.cpp

@@ -114,6 +114,62 @@ namespace llarp::rpc
               auto ftr = result.get_future();
               msg.send_reply(CreateJSONResponse(ftr.get()));
             })
+        .add_request_command(
+            "endpoint",
+            [&](lokimq::Message& msg) {
+              HandleJSONRequest(msg, [r = m_Router](nlohmann::json obj, ReplyFunction_t reply) {
+                if (r->IsServiceNode())
+                {
+                  reply(CreateJSONError("not supported"));
+                  return;
+                }
+                std::string endpoint = "default";
+                std::unordered_set<service::Address> kills;
+                {
+                  const auto itr = obj.find("endpoint");
+                  if (itr != obj.end())
+                    endpoint = itr->get<std::string>();
+                }
+                {
+                  const auto itr = obj.find("kill");
+                  if (itr != obj.end())
+                  {
+                    if (itr->is_array())
+                    {
+                      for (auto kill_itr = itr->begin(); kill_itr != itr->end(); ++kill_itr)
+                      {
+                        if (kill_itr->is_string())
+                          kills.emplace(kill_itr->get<std::string>());
+                      }
+                    }
+                    else if (itr->is_string())
+                    {
+                      kills.emplace(itr->get<std::string>());
+                    }
+                  }
+                }
+                if (kills.empty())
+                {
+                  reply(CreateJSONError("no action taken"));
+                  return;
+                }
+                LogicCall(r->logic(), [r, endpoint, kills, reply]() {
+                  auto ep = r->hiddenServiceContext().GetEndpointByName(endpoint);
+                  if (ep == nullptr)
+                  {
+                    reply(CreateJSONError("no endpoint with name " + endpoint));
+                    return;
+                  }
+                  std::size_t removed = 0;
+                  for (auto kill : kills)
+                  {
+                    removed += ep->RemoveAllConvoTagsFor(std::move(kill));
+                  }
+                  reply(CreateJSONResponse(
+                      "removed " + std::to_string(removed) + " flow" + (removed == 1 ? "" : "s")));
+                });
+              });
+            })
         .add_request_command(
             "exit",
             [&](lokimq::Message& msg) {

llarp/service/address.hpp

@@ -104,4 +104,17 @@ namespace llarp
   }  // namespace service
 }  // namespace llarp
 
+namespace std
+{
+  template <>
+  struct hash<llarp::service::Address>
+  {
+    size_t
+    operator()(const llarp::service::Address& addr) const
+    {
+      return llarp::service::Address::Hash{}(addr);
+    }
+  };
+}  // namespace std
+
 #endif

llarp/service/endpoint.cpp

@@ -345,6 +345,25 @@ namespace llarp
       itr->second.lastUsed = Now();
     }
 
+    size_t
+    Endpoint::RemoveAllConvoTagsFor(service::Address remote)
+    {
+      size_t removed = 0;
+      auto& sessions = Sessions();
+      auto itr = sessions.begin();
+      while (itr != sessions.end())
+      {
+        if (itr->second.remote.Addr() == remote)
+        {
+          itr = sessions.erase(itr);
+          removed++;
+        }
+        else
+          ++itr;
+      }
+      return removed;
+    }
+
     bool
     Endpoint::GetSenderFor(const ConvoTag& tag, ServiceInfo& si) const
     {

llarp/service/endpoint.hpp

@@ -269,6 +269,9 @@ namespace llarp
 
       using PendingBufferQueue = std::deque<PendingBuffer>;
 
+      size_t
+      RemoveAllConvoTagsFor(service::Address remote);
+
       bool
       WantsOutboundSession(const Address&) const override;