Fixed consensus issues with large no. of users and inputs. (#280)

* Fixed issue in consensus contract output  sorting.

* Improved load test script.

* Removed flatbuffer verifier restrictions to support large no. of users and inputs.

* Improved consensus proposal cleanup.

* Restored the earlier proposal cleanup logic.

* Minor fixes.

* const fix.

src/consensus.cpp

@@ -118,6 +118,10 @@ namespace consensus
 
         if (ctx.stage == 0)
         {
+            // Broadcast non-unl proposal (NUP) containing inputs from locally connected users.
+            // This is performed at stage 0, so we can to make sure this happens regardless of whether we are in-sync or not.
+            broadcast_nonunl_proposal();
+
             // Prepare the consensus candidate user inputs that we have accumulated so far. (We receive them periodically via NUPs)
             // The candidate inputs will be included in the stage 0 proposal.
             if (verify_and_populate_candidate_user_inputs(lcl_id.seq_no) == -1)
@@ -161,14 +165,6 @@ namespace consensus
                     LOG_ERROR << "Error occured in Stage 3 consensus execution.";
             }
 
-            if (ctx.stage == 2)
-            {
-                // At end of stage 2, broadcast non-unl proposal (NUP) containing inputs from locally connected users.
-                // This will be captured and verified during every round stage 0.
-                // (We broadcast this at stage 2 in order to give it enough time to reach others before next round stage 0)
-                broadcast_nonunl_proposal();
-            }
-
             // We have finished a consensus stage. Transition or reset stage based on sync status.
 
             if (sync_status == -2)
@@ -319,7 +315,6 @@ namespace consensus
             ctx.candidate_proposals.emplace(proposal.pubkey, std::move(proposal));
         }
 
-        // Prune any outdated proposals.
         auto itr = ctx.candidate_proposals.begin();
         const uint64_t time_now = util::get_epoch_milliseconds();
         while (itr != ctx.candidate_proposals.end())
@@ -610,6 +605,7 @@ namespace consensus
         p.patch_hash = patch_hash;
         p.last_primary_shard_id = last_primary_shard_id;
         p.last_blob_shard_id = last_blob_shard_id;
+        p.output_hash.resize(BLAKE3_OUT_LEN); // Default empty hash.
 
         const uint64_t time_now = util::get_epoch_milliseconds();
 
@@ -753,7 +749,7 @@ namespace consensus
         // If that's the case we should request shards straight away.
         if (ledger::ctx.get_last_primary_shard_id() != majority_primary_shard_id)
         {
-            LOG_DEBUG << "We are not on the consensus ledger, we must request history from a peer.";
+            LOG_INFO << "We are not on the consensus ledger, we must request history from a peer.";
             is_desync = true;
             return true;
         }
@@ -1047,18 +1043,20 @@ namespace consensus
         {
             if (!bufs.outputs.empty())
             {
-                std::vector<std::string_view> vect;
-                // Adding public key.
-                vect.push_back(pubkey);
-                // Only using message to generate hash for output messages. Length is not needed.
-                for (sc::contract_output &output : bufs.outputs)
-                {
-                    vect.push_back(output.message);
-                }
+                // Generate hash of all sorted outputs combined with user pubkey.
+
+                std::vector<std::string_view> vect;
+                for (sc::contract_output &output : bufs.outputs)
+                    vect.push_back(output.message);
+
+                // We sort all outputs so every node calculates the final hash the same way.
+                std::sort(vect.begin(), vect.end());
+
+                // Adding user public key.
+                vect.push_back(pubkey);
 
-                const std::string hash = crypto::get_hash(vect);
                 ctx.generated_user_outputs.try_emplace(
-                    std::move(hash),
+                    crypto::get_hash(vect),
                     generated_user_output(pubkey, std::move(bufs.outputs)));
             }
         }

src/msg/fbuf/p2pmsg_conversion.cpp

@@ -31,7 +31,7 @@ namespace msg::fbuf::p2pmsg
         const size_t buf_size = message.size();
 
         // Verify container message using flatbuffer verifier
-        flatbuffers::Verifier verifier(buf, buf_size, 16, 100);
+        flatbuffers::Verifier verifier(buf, buf_size);
         return VerifyP2PMsgBuffer(verifier);
     }
 

src/util/h32.cpp

@@ -61,10 +61,15 @@ namespace util
 
     std::ostream &operator<<(std::ostream &output, const h32 &h)
     {
+        const std::ios_base::fmtflags stream_flags(output.flags());
+        output << std::hex;
+
         const uint8_t *buf = reinterpret_cast<const uint8_t *>(&h);
         for (int i = 0; i < 5; i++) // Only print first 5 bytes in hex.
-            output << std::hex << std::setfill('0') << std::setw(2) << (int)buf[i];
+            output << std::setfill('0') << std::setw(2) << (int)buf[i];
 
+        // Reset the ostream flags because we set std::hex at the begining.
+        output.flags(stream_flags);
         return output;
     }