add local publish intro message to send reply down path

llarp/dht/context.cpp

@@ -118,12 +118,17 @@ namespace llarp
       RelayRequestForPath(const llarp::PathID_t& localPath,
                           const IMessage& msg) override;
 
+      /// send introset to peer as R/S
+      void
+      PropagateLocalIntroSet(const PathID_t& from, uint64_t txid,
+                             const service::EncryptedIntroSet& introset,
+                             const Key_t& tellpeer, uint64_t relayOrder);
+
       /// send introset to peer from source with S counter and excluding peers
       void
       PropagateIntroSetTo(const Key_t& from, uint64_t txid,
                           const service::EncryptedIntroSet& introset,
-                          const Key_t& tellpeer, bool relayed,
-                          uint64_t relayOrder);
+                          const Key_t& tellpeer, uint64_t relayOrder);
 
       /// initialize dht context and explore every exploreInterval milliseconds
       void
@@ -536,14 +541,26 @@ namespace llarp
     void
     Context::PropagateIntroSetTo(const Key_t& from, uint64_t txid,
                                  const service::EncryptedIntroSet& introset,
-                                 const Key_t& tellpeer, bool relayed,
-                                 uint64_t relayOrder)
+                                 const Key_t& tellpeer, uint64_t relayOrder)
     {
       const TXOwner asker(from, txid);
       const TXOwner peer(tellpeer, ++ids);
       _pendingIntrosetLookups.NewTX(
           peer, asker, asker,
-          new PublishServiceJob(asker, introset, this, relayed, relayOrder));
+          new PublishServiceJob(asker, introset, this, relayOrder));
+    }
+
+    void
+    Context::PropagateLocalIntroSet(const PathID_t& from, uint64_t txid,
+                                    const service::EncryptedIntroSet& introset,
+                                    const Key_t& tellpeer, uint64_t relayOrder)
+    {
+      const TXOwner asker(OurKey(), txid);
+      const TXOwner peer(tellpeer, ++ids);
+      _pendingIntrosetLookups.NewTX(
+          peer, asker, peer,
+          new LocalPublishServiceJob(peer, from, txid, introset, this,
+                                     relayOrder));
     }
 
     void

llarp/dht/context.hpp

@@ -89,12 +89,17 @@ namespace llarp
       virtual bool
       RelayRequestForPath(const PathID_t& localPath, const IMessage& msg) = 0;
 
+      /// send introset to peer from source with S counter and excluding peers
+      virtual void
+      PropagateLocalIntroSet(const PathID_t& path, uint64_t sourceTX,
+                             const service::EncryptedIntroSet& introset,
+                             const Key_t& peer, uint64_t relayOrder) = 0;
+
       /// send introset to peer from source with S counter and excluding peers
       virtual void
       PropagateIntroSetTo(const Key_t& source, uint64_t sourceTX,
                           const service::EncryptedIntroSet& introset,
-                          const Key_t& peer, bool relayed,
-                          uint64_t relayOrder) = 0;
+                          const Key_t& peer, uint64_t relayOrder) = 0;
 
       virtual void
       Init(const Key_t& us, AbstractRouter* router) = 0;

llarp/dht/messages/pubintro.cpp

@@ -112,8 +112,14 @@ namespace llarp
         else
         {
           llarp::LogInfo("propagating to peer ", index);
-
-          dht.PropagateIntroSetTo(From, txID, introset, peer, false, 0);
+          if(relayed)
+          {
+            dht.PropagateLocalIntroSet(pathID, txID, introset, peer, 0);
+          }
+          else
+          {
+            dht.PropagateIntroSetTo(From, txID, introset, peer, 0);
+          }
         }
       };
 

llarp/dht/publishservicejob.cpp

@@ -3,17 +3,19 @@
 #include <dht/context.hpp>
 #include <dht/messages/pubintro.hpp>
 #include <dht/messages/gotintro.hpp>
-#include <utility>
+#include <path/path_context.hpp>
+#include <routing/dht_message.hpp>
+#include <router/abstractrouter.hpp>
 
+#include <utility>
 namespace llarp
 {
   namespace dht
   {
     PublishServiceJob::PublishServiceJob(
         const TXOwner &asker, const service::EncryptedIntroSet &introset_,
-        AbstractContext *ctx, bool relayed_, uint64_t relayOrder_)
+        AbstractContext *ctx, uint64_t relayOrder_)
         : TX< TXOwner, service::EncryptedIntroSet >(asker, asker, ctx)
-        , relayed(relayed_)
         , relayOrder(relayOrder_)
         , introset(introset_)
     {
@@ -37,7 +39,7 @@ namespace llarp
     {
       parent->DHTSendTo(
           peer.node.as_array(),
-          new PublishIntroMessage(introset, peer.txid, relayed, relayOrder));
+          new PublishIntroMessage(introset, peer.txid, false, relayOrder));
     }
 
     void
@@ -47,5 +49,38 @@ namespace llarp
                         new GotIntroMessage({introset}, whoasked.txid));
     }
 
+    LocalPublishServiceJob::LocalPublishServiceJob(
+        const TXOwner &peer, const PathID_t &fromID, uint64_t _txid,
+        const service::EncryptedIntroSet &introset, AbstractContext *ctx,
+        uint64_t relayOrder)
+        : PublishServiceJob(peer, introset, ctx, relayOrder)
+        , localPath(fromID)
+        , txid(_txid)
+    {
+    }
+
+    void
+    LocalPublishServiceJob::SendReply()
+    {
+      auto path = parent->GetRouter()->pathContext().GetByUpstream(
+          parent->OurKey().as_array(), localPath);
+      if(!path)
+      {
+        llarp::LogWarn(
+            "did not send reply for relayed dht request, no such local path "
+            "for pathid=",
+            localPath);
+        return;
+      }
+      routing::DHTMessage msg;
+      msg.M.emplace_back(new GotIntroMessage({introset}, whoasked.txid));
+      if(!path->SendRoutingMessage(msg, parent->GetRouter()))
+      {
+        llarp::LogWarn(
+            "failed to send routing message when informing result of dht "
+            "request, pathid=",
+            localPath);
+      }
+    }
   }  // namespace dht
 }  // namespace llarp

llarp/dht/publishservicejob.hpp

@@ -14,14 +14,12 @@ namespace llarp
   {
     struct PublishServiceJob : public TX< TXOwner, service::EncryptedIntroSet >
     {
-      bool relayed;
       uint64_t relayOrder;
       service::EncryptedIntroSet introset;
 
       PublishServiceJob(const TXOwner &asker,
                         const service::EncryptedIntroSet &introset,
-                        AbstractContext *ctx, bool relayed,
-                        uint64_t relayOrder);
+                        AbstractContext *ctx, uint64_t relayOrder);
 
       bool
       Validate(const service::EncryptedIntroSet &introset) const override;
@@ -29,6 +27,19 @@ namespace llarp
       void
       Start(const TXOwner &peer) override;
 
+      virtual void
+      SendReply() override;
+    };
+
+    struct LocalPublishServiceJob : public PublishServiceJob
+    {
+      PathID_t localPath;
+      uint64_t txid;
+      LocalPublishServiceJob(const TXOwner &peer, const PathID_t &fromID,
+                             uint64_t txid,
+                             const service::EncryptedIntroSet &introset,
+                             AbstractContext *ctx, uint64_t relayOrder);
+
       void
       SendReply() override;
     };

test/dht/mock_context.hpp

@@ -59,10 +59,14 @@ namespace llarp
       MOCK_CONST_METHOD2(GetRCFromNodeDB,
                          bool(const dht::Key_t& k, RouterContact& rc));
 
-      MOCK_METHOD6(PropagateIntroSetTo,
+      MOCK_METHOD5(PropagateIntroSetTo,
                    void(const dht::Key_t& source, uint64_t sourceTX,
                         const service::EncryptedIntroSet& introset,
-                        const dht::Key_t& peer, bool relayed, uint64_t relayOrder));
+                        const dht::Key_t& peer, uint64_t relayOrder));
+      MOCK_METHOD5(PropagateLocalIntroSet,
+                   void(const PathID_t& source, uint64_t sourceTX,
+                        const service::EncryptedIntroSet& introset,
+                        const dht::Key_t& peer, uint64_t relayOrder));
 
       MOCK_METHOD2(Init,
                    void(const dht::Key_t&, AbstractRouter*));