Merge pull request #1541 from majestrate/lns-consensus-2021-02-19

lns lookup consensus

llarp/CMakeLists.txt

@@ -201,6 +201,7 @@ add_library(liblokinet
   service/info.cpp
   service/intro_set.cpp
   service/intro.cpp
+  service/lns_tracker.cpp
   service/lookup.cpp
   service/name.cpp
   service/outbound_context.cpp

llarp/handlers/tun.cpp

@@ -354,6 +354,20 @@ namespace llarp
             2s);
       };
 
+      auto ReplyToDNSWhenReady = [ReplyToLokiDNSWhenReady, ReplyToSNodeDNSWhenReady](
+                                     auto addr, auto msg, bool isV6) {
+        if (auto ptr = std::get_if<RouterID>(&addr))
+        {
+          ReplyToSNodeDNSWhenReady(*ptr, msg, isV6);
+          return;
+        }
+        if (auto ptr = std::get_if<service::Address>(&addr))
+        {
+          ReplyToLokiDNSWhenReady(*ptr, msg, isV6);
+          return;
+        }
+      };
+
       auto ReplyToLokiSRVWhenReady = [self = this, reply = reply](
                                          service::Address addr, auto msg) -> bool {
         using service::Address;
@@ -483,10 +497,10 @@ namespace llarp
         }
         else if (service::NameIsValid(lnsName))
         {
-          return LookupNameAsync(lnsName, [msg, lnsName, reply](auto maybe) mutable {
+          LookupNameAsync(lnsName, [msg, lnsName, reply](auto maybe) mutable {
             if (maybe.has_value())
             {
-              msg.AddMXReply(maybe->ToString(), 1);
+              var::visit([&](auto&& value) { msg.AddMXReply(value.ToString(), 1); }, *maybe);
             }
             else
             {
@@ -494,6 +508,7 @@ namespace llarp
             }
             reply(msg);
           });
+          return true;
         }
         else
           msg.AddNXReply();
@@ -617,14 +632,14 @@ namespace llarp
         }
         else if (service::NameIsValid(lnsName))
         {
-          return LookupNameAsync(
+          LookupNameAsync(
               lnsName,
               [msg = std::make_shared<dns::Message>(msg),
                name = Name(),
                lnsName,
                isV6,
                reply,
-               ReplyToLokiDNSWhenReady](auto maybe) {
+               ReplyToDNSWhenReady](auto maybe) {
                 if (not maybe.has_value())
                 {
                   LogWarn(name, " lns name ", lnsName, " not resolved");
@@ -632,9 +647,9 @@ namespace llarp
                   reply(*msg);
                   return;
                 }
-                LogInfo(name, " ", lnsName, " resolved to ", maybe->ToString());
+                ReplyToDNSWhenReady(*maybe, msg, isV6);
-                ReplyToLokiDNSWhenReady(*maybe, msg, isV6);
               });
+          return true;
         }
         else
           msg.AddNXReply();

llarp/rpc/rpc_server.cpp

@@ -294,12 +294,15 @@ namespace llarp::rpc
                           reply(CreateJSONError("we could not find an exit with that name"));
                           return;
                         }
-                        if (maybe->IsZero())
+                        if (auto ptr = std::get_if<service::Address>(&*maybe))
                         {
-                          reply(CreateJSONError("lokinet exit does not exist"));
+                          mapExit(*ptr);
+                        }
+                        else
+                        {
+                          reply(CreateJSONError("lns name resolved to a snode"));
                           return;
                         }
-                        mapExit(*maybe);
                       });
                     }
                     else

llarp/service/endpoint.cpp

@@ -232,12 +232,13 @@ namespace llarp
                   const auto maybe_auth = info.second;
 
                   m_StartupLNSMappings.erase(name);
-
+                  if (auto* addr = std::get_if<service::Address>(&*maybe_addr))
-                  if (maybe_range.has_value())
+                  {
-                    m_ExitMap.Insert(*maybe_range, *maybe_addr);
+                    if (maybe_range.has_value())
-
+                      m_ExitMap.Insert(*maybe_range, *addr);
-                  if (maybe_auth.has_value())
+                    if (maybe_auth.has_value())
-                    SetAuthInfoForEndpoint(*maybe_addr, *maybe_auth);
+                      SetAuthInfoForEndpoint(*addr, *maybe_auth);
+                  }
                 }
               });
         }
@@ -798,22 +799,65 @@ namespace llarp
       return not m_ExitMap.Empty();
     }
 
-    bool
+    void
-    Endpoint::LookupNameAsync(std::string name, std::function<void(std::optional<Address>)> handler)
+    Endpoint::LookupNameAsync(
+        std::string name,
+        std::function<void(std::optional<std::variant<Address, RouterID>>)> handler)
     {
       auto& cache = m_state->nameCache;
       const auto maybe = cache.Get(name);
       if (maybe.has_value())
       {
         handler(maybe);
-        return true;
+        return;
       }
-      auto path = PickRandomEstablishedPath();
-      if (path == nullptr)
-        return false;
       LogInfo(Name(), " looking up LNS name: ", name);
-      auto job = new LookupNameJob(this, GenTXID(), name, handler);
+      path::Path::UniqueEndpointSet_t paths;
-      return job->SendRequestViaPath(path, m_router);
+      ForEachPath([&](auto path) {
+        if (path->IsReady())
+        {
+          paths.insert(path);
+        }
+      });
+      // not enough paths
+      if (paths.size() < 3)
+      {
+        handler(std::nullopt);
+        return;
+      }
+
+      auto maybeInvalidateCache = [handler, &cache, name](auto result) {
+        if (result)
+        {
+          var::visit(
+              [&](auto&& value) {
+                if (value.IsZero())
+                {
+                  result = std::nullopt;
+                }
+              },
+              *result);
+        }
+        if (result)
+        {
+          cache.Put(name, *result);
+        }
+        else
+        {
+          cache.Remove(name);
+        }
+        handler(result);
+      };
+
+      auto resultHandler =
+          m_state->lnsTracker.MakeResultHandler(name, paths.size(), maybeInvalidateCache);
+
+      for (const auto& path : paths)
+      {
+        LogInfo(Name(), " lookup ", name, " from ", path->Endpoint());
+        auto job = new LookupNameJob(this, GenTXID(), name, resultHandler);
+        job->SendRequestViaPath(path, m_router);
+      }
     }
 
     bool
@@ -826,12 +870,6 @@ namespace llarp
 
       // decrypt entry
       const auto maybe = msg->result.Decrypt(itr->second->name);
-
-      if (maybe.has_value())
-      {
-        // put cache entry for result
-        m_state->nameCache.Put(itr->second->name, *maybe);
-      }
       // inform result
       itr->second->HandleNameResponse(maybe);
       lookups.erase(itr);

llarp/service/endpoint.hpp

@@ -21,6 +21,8 @@
 
 #include "auth.hpp"
 
+#include <oxenmq/variant.h>
+
 // minimum time between introset shifts
 #ifndef MIN_SHIFT_INTERVAL
 #define MIN_SHIFT_INTERVAL 5s
@@ -218,8 +220,10 @@ namespace llarp
       bool
       LookupRouterAnon(RouterID router, RouterLookupHandler handler);
 
-      bool
+      void
-      LookupNameAsync(std::string name, std::function<void(std::optional<Address>)> resultHandler);
+      LookupNameAsync(
+          std::string name,
+          std::function<void(std::optional<std::variant<Address, RouterID>>)> resultHandler);
 
       /// called on event loop pump
       virtual void

llarp/service/endpoint_state.hpp

@@ -10,12 +10,15 @@
 #include <llarp/util/compare_ptr.hpp>
 #include <llarp/util/decaying_hashtable.hpp>
 #include <llarp/util/status.hpp>
+#include "lns_tracker.hpp"
 
 #include <memory>
 #include <queue>
 #include <set>
 #include <unordered_map>
 
+#include <oxenmq/variant.h>
+
 namespace llarp
 {
   namespace service
@@ -64,7 +67,10 @@ namespace llarp
 
       OutboundSessions_t m_OutboundSessions;
 
-      util::DecayingHashTable<std::string, Address, std::hash<std::string>> nameCache;
+      util::DecayingHashTable<std::string, std::variant<Address, RouterID>, std::hash<std::string>>
+          nameCache;
+
+      LNSLookupTracker lnsTracker;
 
       bool
       Configure(const NetworkConfig& conf);

llarp/service/lns_tracker.cpp

@@ -0,0 +1,48 @@
+#include "lns_tracker.hpp"
+
+namespace llarp::service
+{
+  std::function<void(std::optional<LNSLookupTracker::Addr_t>)>
+  LNSLookupTracker::MakeResultHandler(
+      std::string name,
+      std::size_t numPeers,
+      std::function<void(std::optional<Addr_t>)> resultHandler)
+  {
+    m_PendingLookups.emplace(name, LookupInfo{numPeers, resultHandler});
+    return [name, this](std::optional<Addr_t> found) {
+      auto itr = m_PendingLookups.find(name);
+      if (itr == m_PendingLookups.end())
+        return;
+      itr->second.HandleOneResult(found);
+      if (itr->second.IsDone())
+        m_PendingLookups.erase(itr);
+    };
+  }
+
+  bool
+  LNSLookupTracker::LookupInfo::IsDone() const
+  {
+    return m_ResultsGotten == m_ResultsNeeded;
+  }
+
+  void
+  LNSLookupTracker::LookupInfo::HandleOneResult(std::optional<Addr_t> result)
+  {
+    if (result)
+    {
+      m_CurrentValues.insert(*result);
+    }
+    m_ResultsGotten++;
+    if (IsDone())
+    {
+      if (m_CurrentValues.size() == 1)
+      {
+        m_HandleResult(*m_CurrentValues.begin());
+      }
+      else
+      {
+        m_HandleResult(std::nullopt);
+      }
+    }
+  }
+}  // namespace llarp::service

llarp/service/lns_tracker.hpp

@@ -0,0 +1,53 @@
+#pragma once
+
+#include <functional>
+#include <optional>
+#include <unordered_map>
+#include <unordered_set>
+#include <string>
+
+#include "address.hpp"
+#include <llarp/router_id.hpp>
+#include <oxenmq/variant.h>
+
+namespace llarp::service
+{
+  /// tracks and manages consensus of lns names we fetch from the network
+  class LNSLookupTracker
+  {
+   public:
+    using Addr_t = std::variant<Address, RouterID>;
+
+   private:
+    struct LookupInfo
+    {
+      std::unordered_set<Addr_t> m_CurrentValues;
+      std::function<void(std::optional<Addr_t>)> m_HandleResult;
+      std::size_t m_ResultsGotten = 0;
+      std::size_t m_ResultsNeeded;
+
+      LookupInfo(std::size_t wantResults, std::function<void(std::optional<Addr_t>)> resultHandler)
+          : m_HandleResult{std::move(resultHandler)}, m_ResultsNeeded{wantResults}
+      {}
+
+      bool
+      IsDone() const;
+
+      void
+      HandleOneResult(std::optional<Addr_t> result);
+    };
+
+    std::unordered_map<std::string, LookupInfo> m_PendingLookups;
+
+   public:
+    /// make a function that will handle consensus of an lns request
+    /// name is the name we are requesting
+    /// numPeers is the number of peers we asked
+    /// resultHandler is a function that we are wrapping that will handle the final result
+    std::function<void(std::optional<Addr_t>)>
+    MakeResultHandler(
+        std::string name,
+        std::size_t numPeers,
+        std::function<void(std::optional<Addr_t>)> resultHandler);
+  };
+}  // namespace llarp::service

llarp/util/decaying_hashtable.hpp

@@ -17,6 +17,7 @@ namespace llarp::util
       EraseIf([&](const auto& item) { return item.second.second + m_CacheInterval <= now; });
     }
 
+    /// return if we have this value by key
     bool
     Has(const Key_t& k) const
     {
@@ -33,6 +34,7 @@ namespace llarp::util
       return m_Values.try_emplace(std::move(key), std::make_pair(std::move(value), now)).second;
     }
 
+    /// get value by key
     std::optional<Value_t>
     Get(Key_t k) const
     {
@@ -42,6 +44,13 @@ namespace llarp::util
       return itr->second.first;
     }
 
+    /// explicit remove an item from the cache by key
+    void
+    Remove(const Key_t& key)
+    {
+      m_Values.erase(key);
+    }
+
    private:
     template <typename Predicate_t>
     void