limits and have clients use ephemeral identity

include/llarp/link/session.hpp

@@ -46,6 +46,9 @@ namespace llarp
 
     /// handle a valid LIM
     std::function< bool(const LinkIntroMessage *msg) > GotLIM;
+
+    /// send queue current blacklog
+    std::function< size_t(void) > SendQueueBacklog;
   };
 }  // namespace llarp
 

llarp/link/server.cpp

@@ -161,14 +161,20 @@ namespace llarp
     Lock l(m_AuthedLinksMutex);
     auto range = m_AuthedLinks.equal_range(remote);
     auto itr   = range.first;
-    // TODO: random selection
+    // pick lowest backlog session
+    size_t min      = std::numeric_limits< size_t >::max();
+    ILinkSession* s = nullptr;
     while(itr != range.second)
     {
-      if(itr->second->SendMessageBuffer(buf))
-        return true;
+      auto backlog = itr->second->SendQueueBacklog();
+      if(backlog < min)
+      {
+        s   = itr->second.get();
+        min = backlog;
+      }
       ++itr;
     }
-    return false;
+    return s && s->SendMessageBuffer(buf);
   }
 
   bool

llarp/link/utp.cpp

@@ -33,6 +33,9 @@ namespace llarp
 
     constexpr size_t MaxSend = 64;
 
+    /// maximum size for send queue for a session before we drop
+    constexpr size_t MaxSendQueueSize = 128;
+
     typedef llarp::AlignedBuffer< MAX_LINK_MSG_SIZE > MessageBuffer;
 
     struct LinkLayer;
@@ -175,6 +178,8 @@ namespace llarp
       bool
       QueueWriteBuffers(llarp_buffer_t buf)
       {
+        if(sendq.size() >= MaxSendQueueSize)
+          return false;
         llarp::LogDebug("write ", buf.sz, " bytes to ", remoteAddr);
         lastActive  = llarp_time_now_ms();
         size_t sz   = buf.sz;
@@ -568,6 +573,8 @@ namespace llarp
       remoteTransportPubKey.Zero();
       recvMsgOffset = 0;
 
+      SendQueueBacklog = [&]() -> size_t { return sendq.size(); };
+
       SendKeepAlive = [&]() -> bool {
         if(sendq.size() == 0 && state == eSessionReady)
         {

llarp/router.cpp

@@ -152,6 +152,8 @@ llarp_router::PersistSessionUntil(const llarp::RouterID &remote,
   }
 }
 
+constexpr size_t MaxPendingSendQueueSize = 8;
+
 bool
 llarp_router::SendToOrQueue(const llarp::RouterID &remote,
                             const llarp::ILinkMessage *msg)
@@ -181,7 +183,15 @@ llarp_router::SendToOrQueue(const llarp::RouterID &remote,
     return false;
   // queue buffer
   auto &q = outboundMessageQueue[remote];
-  buf.sz  = buf.cur - buf.base;
+
+  if(q.size() >= MaxPendingSendQueueSize)
+  {
+    llarp::LogWarn("tried to queue a message to ", remote,
+                   " but the queue is full so we drop it like it's hawt");
+    return false;
+  }
+
+  buf.sz = buf.cur - buf.base;
   q.emplace(buf.sz);
   memcpy(q.back().data(), buf.base, buf.sz);
 
@@ -726,6 +736,18 @@ llarp_router::Run()
   }
   else
   {
+    // we are a client
+    // regenerate keys and resign rc before everything else
+    crypto.identity_keygen(identity);
+    crypto.encryption_keygen(encryption);
+    _rc.pubkey = llarp::seckey_topublic(identity);
+    _rc.enckey = llarp::seckey_topublic(encryption);
+    if(!_rc.Sign(&crypto, identity))
+    {
+      llarp::LogError("failed to regenerate keys and sign RC");
+      return;
+    }
+
     // delayed connect all for clients
     uint64_t delay = ((llarp_randint() % 10) * 500) + 500;
     llarp_logic_call_later(logic, {delay, this, &ConnectAll});

llarp/service/endpoint.cpp

@@ -912,6 +912,7 @@ namespace llarp
             llarp::LogError("no path found");
             return false;
           }
+          // TODO: check expiration of our end
           ProtocolMessage m(f.T);
           m.proto      = t;
           m.introReply = p->intro;
@@ -933,6 +934,8 @@ namespace llarp
       }
       if(HasPathToService(remote))
       {
+        llarp::LogDebug(Name(), " has session to ", remote, " sending ",
+                        data.sz, " bytes");
         m_RemoteSessions[remote]->AsyncEncryptAndSendTo(data, t);
         return true;
       }