Udht fixes

data/config.ini

@@ -35,6 +35,8 @@ wallet-public-address=
 external-address=
 jump-node-coefficient=
 redirect-timeout-ms=
+;;duplicate-filter-timeout-ms optinal parameter, 30000 by default, time interval to consider the same requests as duplicates; 0 means disable the filter
+duplicate-filter-timeout-ms=
 
 [stun]
 enabled=false

include/lib/graft/connection.h

@@ -220,7 +220,7 @@ class XorolFilter;
 class HttpConnectionManager final : public ConnectionManager
 {
 public:
-    HttpConnectionManager();
+    HttpConnectionManager(uint expire_ms);
 
     void bind(Looper& looper) override;
 

include/lib/graft/serveropts.h

@@ -38,9 +38,9 @@ struct ConfigOpts
     int log_trunc_to_size;
     std::vector<std::string> graftlet_dirs;
     int lru_timeout_ms;
+    int duplicate_filter_timeout_ms;
     IPFilterOpts ipfilter;
     CommonOpts common;
-    bool duplicate_filter_enabled = true;
 
     void check_asserts() const
     {
@@ -51,6 +51,7 @@ struct ConfigOpts
         assert(0 < workers_expelling_interval_ms);
         assert(0 < timer_poll_interval_ms);
         assert(0 < lru_timeout_ms);
+        assert(0 <= duplicate_filter_timeout_ms);
         assert(ipfilter.requests_per_sec == 0 || 0 < ipfilter.window_size_sec);
     }
 };

include/supernode/requests/redirect.h

@@ -26,6 +26,7 @@ GRAFT_DEFINE_JSON_RPC_REQUEST(RegisterSupernodeJsonRpcRequest, RegisterSupernode
 GRAFT_DEFINE_IO_STRUCT(UDHTInfoItem,
                        (std::string, id),
                        (std::string, ip_port),
+                       (bool, active),
                        (std::string, expiration_time),
                        (uint32_t, broadcast_count),
                        (double, avg_hop_ip_broadcast),

src/lib/graft/connection.cpp

@@ -278,7 +278,7 @@ ConnectionManager* ConnectionBase::getConMgr(const ConnectionManager::Proto& pro
 
 void ConnectionBase::initConnectionManagers()
 {
-    std::unique_ptr<HttpConnectionManager> httpcm = std::make_unique<HttpConnectionManager>();
+    std::unique_ptr<HttpConnectionManager> httpcm = std::make_unique<HttpConnectionManager>(getCopts().duplicate_filter_timeout_ms);
     auto res1 = m_conManagers.emplace(httpcm->getProto(), std::move(httpcm));
     assert(res1.second);
     std::unique_ptr<CoapConnectionManager> coapcm = std::make_unique<CoapConnectionManager>();
@@ -467,14 +467,13 @@ CoapConnectionManager* CoapConnectionManager::from_accepted(mg_connection* cn)
 
 class XorolFilter : public detail::ExpiringSetT<uint64_t>
 {
-    static constexpr int lifetime = 30000;
 public:
-    XorolFilter() : detail::ExpiringSetT<uint64_t>(lifetime) { }
+    XorolFilter(uint expire_ms) : detail::ExpiringSetT<uint64_t>(expire_ms) { }
 };
 
-HttpConnectionManager::HttpConnectionManager()
+HttpConnectionManager::HttpConnectionManager(uint expire_ms)
     : ConnectionManager("HTTP")
-    , m_xorolFilter(std::make_unique<XorolFilter>())
+    , m_xorolFilter(std::make_unique<XorolFilter>(expire_ms))
 {
 }
 
@@ -495,10 +494,10 @@ void HttpConnectionManager::ev_handler_http(mg_connection *client, int ev, void
         conBase->getLooper().runtimeSysInfo().count_http_req_bytes_raw(hm->message.len);
 
         const ConfigOpts& opts = conBase->getLooper().getCopts();
-        if(opts.duplicate_filter_enabled)
+        if(opts.duplicate_filter_timeout_ms)
         {//calculate hash value of body, and check if it is in XorolFilter
             mg_str& body = hm->body;
-            uint16_t xr = graft::utils::xorol(reinterpret_cast<const uint8_t*>(body.p), body.len);
+            uint64_t xr = graft::utils::xorol(reinterpret_cast<const uint8_t*>(body.p), body.len);
 
             HttpConnectionManager* httpCM = dynamic_cast<HttpConnectionManager*>( conBase->getConMgr("HTTP") );
             assert(httpCM);
@@ -507,7 +506,7 @@ void HttpConnectionManager::ev_handler_http(mg_connection *client, int ev, void
             {
                 conBase->getLooper().runtimeSysInfo().count_http_request_unrouted();
 
-                LOG_PRINT_CLN(2,client,"Duplicating hash found '") << std::ios::hex << xr << std::ios::dec << "'; closing connection";
+                LOG_PRINT_CLN(3,client,"Duplicating hash found '") << std::ios::hex << xr << std::ios::dec << "'; closing connection";
                 mg_http_send_error(client, 409, "duplicate request found");
                 client->flags |= MG_F_SEND_AND_CLOSE;
                 break;

src/lib/graft/task.cpp

@@ -554,7 +554,7 @@ bool TaskManager::tryProcessReadyJob()
     ++m_cntJobDone;
     BaseTaskPtr bt = gj->getTask();
 
-    LOG_PRINT_RQS_BT(2,bt,"worker_action completed with result " << bt->getStrStatus());
+    LOG_PRINT_RQS_BT(3,bt,"worker_action completed with result " << bt->getStrStatus());
     m_stateMachine->dispatch(bt, StateMachine::State::WORKER_ACTION_DONE);
     return true;
 }

src/supernode/requests/redirect.cpp

@@ -44,12 +44,14 @@ namespace graft::supernode::request {
 namespace
 {
 
-const std::chrono::system_clock::duration time_wnd = std::chrono::hours(12);
+using Clock = std::chrono::system_clock;
+
+const Clock::duration time_wnd = std::chrono::hours(12);
 
 struct UdhtStates
 {
     //announce expire time, hops left
-    using ContTimeHop = std::deque<std::pair<std::chrono::system_clock::time_point, uint64_t>>;
+    using ContTimeHop = std::deque<std::pair<Clock::time_point, uint64_t>>;
     ContTimeHop announces;
     ContTimeHop redirects;
 };
@@ -58,7 +60,7 @@ class UdhtInfo
 {
     void chop_old(UdhtStates* item = nullptr)
     {
-        auto edge = std::chrono::system_clock::now() - time_wnd;
+        auto edge = Clock::now() - time_wnd;
         auto chop = [&edge](UdhtStates::ContTimeHop& cont)->void
         {
             auto v = std::make_pair(edge, (uint64_t)0);
@@ -94,7 +96,7 @@ class UdhtInfo
         case Redirects: cont = &item.redirects; break;
         default: assert(false);
         }
-        cont->push_front(std::make_pair(std::chrono::system_clock::now(), hops));
+        cont->push_front(std::make_pair(Clock::now(), hops));
     }
 
     void chop_all_old()
@@ -106,9 +108,9 @@ class UdhtInfo
         }
     }
 
-    static std::string local_time(const std::chrono::system_clock::time_point& tp)
+    static std::string local_time(const Clock::time_point& tp)
     {
-        auto tm = std::chrono::system_clock::to_time_t(tp);
+        auto tm = Clock::to_time_t(tp);
 
         std::ostringstream ss;
         ss << std::put_time(std::localtime(&tm), "%F %T");
@@ -717,29 +719,38 @@ graft::Status onGetUDHTInfo(const graft::Router::vars_t& vars, const graft::Inpu
         res.url = (std::string)ctx.global["supernode_url"];
         res.redirect_uri = "/redirect_broadcast";
 
-        Id2IpShared id2ip = ctx.global["ID:IP:port map"];
+        //make copy of "ID:IP:port map"
+        Id2Ip id2ip;
+        ctx.global.apply<Id2IpShared>("ID:IP:port map", [&id2ip](Id2IpShared& ptr)->bool
+        {
+            id2ip = *ptr;
+        });
+
+        uint32_t redirect_timeout_ms = ctx.global["redirect_timeout_ms"];
 
         ctx.global.apply<UdhtInfoShared>("UdhtInfo", [&](UdhtInfoShared& info)->bool
         {
             info->chop_all_old();
 
+            Clock::time_point tp_now = Clock::now();
+
             // fill res.items
             for(auto& it : info->id2states)
             {
                 std::string id = it.first;
                 UdhtStates states = it.second;
-                auto it_ip = id2ip->find(id);
-                assert(it_ip != id2ip->end());
-                if(it_ip == id2ip->end()) continue;
+                auto it_ip = id2ip.find(id);
+                assert(it_ip != id2ip.end());
+                if(it_ip == id2ip.end()) continue;
                 if(states.announces.empty()) continue;
 
                 UDHTInfoItem res_item;
                 res_item.id = it_ip->first;
                 res_item.ip_port = it_ip->second;
                 {//res_item.expiration_time
                     auto& tp = states.announces.front().first;
-                    uint32_t redirect_timeout_ms = ctx.global["redirect_timeout_ms"];
                     tp += std::chrono::milliseconds( redirect_timeout_ms );
+                    res_item.active = (tp_now <= tp);
                     res_item.expiration_time = UdhtInfo::local_time(tp);
                 }
                 res_item.broadcast_count = states.announces.size();

src/supernode/server.cpp

@@ -532,6 +532,7 @@ bool GraftServer::initConfigOption(int argc, const char** argv, ConfigOpts& conf
     configOpts.workers_expelling_interval_ms = server_conf.get<int>("workers-expelling-interval-ms", 1000);
     configOpts.upstream_request_timeout = server_conf.get<double>("upstream-request-timeout");
     configOpts.lru_timeout_ms = server_conf.get<int>("lru-timeout-ms");
+    configOpts.duplicate_filter_timeout_ms = server_conf.get<int>("duplicate-filter-timeout-ms", 30000);
     configOpts.common.data_dir = server_conf.get<std::string>("data-dir");
     configOpts.common.wallet_public_address = server_conf.get<std::string>("wallet-public-address", "");
     configOpts.common.testnet = server_conf.get<bool>("testnet", false);

test/fixture.h

@@ -29,7 +29,7 @@ class GSTest : public graft::GraftServer
 protected:
     virtual bool initConfigOption(int argc, const char** argv, graft::ConfigOpts& configOpts) override
     {
-        configOpts.duplicate_filter_enabled = false;
+        configOpts.duplicate_filter_timeout_ms = 0;
         if(m_ignoreInitConfig) return true; //prevents loading parameters from command line and config.ini
         return graft::GraftServer::initConfigOption(argc, argv, configOpts);
     }