Improved user inputs handling in consensus (#46)

Updated flatbuffer proposal raw_inputs, raw_outputs data structure.
Improved user inputs handling in consensus.

src/usr/user_session_handler.cpp

@@ -48,7 +48,7 @@ void user_session_handler::on_connect(sock::socket_session<user_outbound_message
     session->init_uniqueid();
 
     // Create an entry in pending_challenges for later tracking upon challenge response.
-    usr::pending_challenges[session->uniqueid] = challengehex;
+    usr::pending_challenges.try_emplace(session->uniqueid, challengehex);
 
     user_outbound_message outmsg(std::move(msgstr));
     session->send(std::move(outmsg));
@@ -95,7 +95,7 @@ void user_session_handler::on_message(
 
                     session->flags.reset(util::SESSION_FLAG::USER_CHALLENGE_ISSUED); // Clear challenge-issued flag
                     session->flags.set(util::SESSION_FLAG::USER_AUTHED);             // Set the user-authed flag
-                    usr::add_user(session, userpubkey);                               // Add the user to the global authed user list
+                    usr::add_user(session, userpubkey);                              // Add the user to the global authed user list
                     usr::pending_challenges.erase(session->uniqueid);                // Remove the stored challenge
 
                     LOG_INFO << "User connection " << session->uniqueid << " authenticated. Public key "
@@ -125,10 +125,13 @@ void user_session_handler::on_message(
             // This is an authed user.
             usr::connected_user &user = itr->second;
 
-            //Append the bytes into connected user input buffer.
-            user.inbuffer.append(message);
+            {
+                std::lock_guard<std::mutex> lock(users_mutex);
+                //Add to the hashed input buffer list.
+                user.inputs.push_back(util::hash_buffer(message, user.pubkey));
+            }
 
-            LOG_DBG << "Collected " << user.inbuffer.length() << " bytes from user";
+            LOG_DBG << "Collected " << message.length() << " bytes from user";
             return;
         }
     }

src/usr/usr.cpp

@@ -31,13 +31,13 @@ std::mutex users_mutex; // Mutex for users access race conditions.
  * This is used for pubkey duplicate checks as well.
  * Map key: User binary pubkey
  */
-std::unordered_map<std::string, std::string> sessionids;
+std::unordered_map<std::string, const std::string> sessionids;
 
 /**
  * Keep track of verification-pending challenges issued to newly connected users.
  * Map key: User socket session id (<ip:port>)
  */
-std::unordered_map<std::string, std::string> pending_challenges;
+std::unordered_map<std::string, const std::string> pending_challenges;
 
 /**
  * User session handler instance. This instance's methods will be fired for any user socket activity.
@@ -152,7 +152,10 @@ int verify_user_challenge_response(std::string &extracted_pubkeyhex, std::string
 {
     // We load response raw bytes into json document.
     rapidjson::Document d;
-    d.Parse(response.data());
+
+    // Because we project the response message directly from the binary socket buffer in a zero-copy manner, the response
+    // string is not null terminated. 'kParseStopWhenDoneFlag' avoids rapidjson error in this case.
+    d.Parse<rapidjson::kParseStopWhenDoneFlag>(response.data());
     if (d.HasParseError())
     {
         LOG_INFO << "Challenge response json parsing failed.";
@@ -226,7 +229,7 @@ int add_user(sock::socket_session<user_outbound_message> *session, const std::st
     }
 
     // Populate sessionid map so we can lookup by user pubkey.
-    sessionids[pubkey] = sessionid;
+    sessionids.try_emplace(pubkey, sessionid);
 
     return 0;
 }
@@ -254,7 +257,7 @@ int remove_user(const std::string &sessionid)
         std::lock_guard<std::mutex> lock(users_mutex);
         sessionids.erase(user.pubkey);
     }
-    
+
     users.erase(itr);
     return 0;
 }
@@ -268,7 +271,6 @@ void start_listening()
     auto address = net::ip::make_address(conf::cfg.listenip);
     sess_opts.max_message_size = conf::cfg.pubmaxsize;
 
-
     std::make_shared<sock::socket_server<user_outbound_message>>(
         ioc,
         ctx,

src/usr/usr.hpp

@@ -4,6 +4,7 @@
 #include <cstdio>
 #include <string_view>
 #include <unordered_map>
+#include <list>
 #include <mutex>
 #include "../util.hpp"
 #include "../sock/socket_session.hpp"
@@ -22,10 +23,10 @@ namespace usr
 struct connected_user
 {
     // User binary public key
-    std::string pubkey;
+    const std::string pubkey;
 
-    // Holds the unprocessed user input collected from websocket.
-    std::string inbuffer;
+    // Holds the unprocessed user inputs (and the hashes) collected from websocket.
+    std::list<util::hash_buffer> inputs;
 
     // Holds the websocket session of this user.
     // We don't need to own the session object since the lifetime of user and session are coupled.
@@ -35,9 +36,9 @@ struct connected_user
      * @param _pubkey The public key of the user in binary format.
      */
     connected_user(sock::socket_session<user_outbound_message> *_session, std::string_view _pubkey)
+        : pubkey(_pubkey)
     {
         session = _session;
-        pubkey = _pubkey;
     }
 };
 
@@ -52,12 +53,12 @@ extern std::mutex users_mutex; // Mutex for users access race conditions.
  * Keep track of verification-pending challenges issued to newly connected users.
  * Map key: User socket session id (<ip:port>)
  */
-extern std::unordered_map<std::string, std::string> sessionids;
+extern std::unordered_map<std::string, const std::string> sessionids;
 
 /**
  * Keep track of verification-pending challenges issued to newly connected users.
  */
-extern std::unordered_map<std::string, std::string> pending_challenges;
+extern std::unordered_map<std::string, const std::string> pending_challenges;
 
 int init();