Merge pull request #842 from skalenetwork/develop

Pull changes from develop

.github/workflows/build.yml

@@ -4,14 +4,17 @@ jobs:
   build:
     runs-on: ubuntu-20.04
     steps:
+    - name: Cancel Previous Runs
+      uses: styfle/[email protected]
+      with:
+        access_token: ${{ github.token }}
 
     - name: checkout
       uses: actions/checkout@v1
 
     - name: submodule update
       run: git submodule update --init --recursive
 
-
     - name: ls
       run: ls
 

.github/workflows/dockerimage.yml

@@ -4,6 +4,12 @@ jobs:
   build:
     runs-on: ubuntu-20.04
     steps:
+
+      - name: Cancel Previous Runs
+        uses: styfle/[email protected]
+        with:
+          access_token: ${{ github.token }}
+
       - name: Free disk space
         run: |
           sudo apt-get remove -yq cmake libjsoncpp-dev aria2 ansible azure-cli shellcheck rpm xorriso zsync \

Consensust.cpp

@@ -94,8 +94,9 @@ uint64_t Consensust::getRunningTimeS() {
     if ( runningTimeS == 0 ) {
         auto env = getenv( "TEST_TIME_S" );
 
-        if ( env != NULL ) {
-            runningTimeS = strtoul( env, NULL, 10 );
+        if ( env ) {
+            runningTimeS = strtoul( env, nullptr, 10 );
+            CHECK_STATE(runningTimeS);
         } else {
             runningTimeS = DEFAULT_RUNNING_TIME_S;
         }

catchup/server/CatchupServerAgent.cpp

@@ -174,7 +174,7 @@ void CatchupServerAgent::processNextAvailableConnection(
 
 
 ptr< vector< uint8_t > > CatchupServerAgent::createResponseHeaderAndBinary(
-        const ptr< ServerConnection >&, nlohmann::json _jsonRequest,
+        const ptr< ServerConnection >& _connection, nlohmann::json _jsonRequest,
         const ptr< Header >& _responseHeader ) {
     CHECK_ARGUMENT( _responseHeader );
 
@@ -197,7 +197,7 @@ ptr< vector< uint8_t > > CatchupServerAgent::createResponseHeaderAndBinary(
         ptr< vector< uint8_t > > serializedBinary = nullptr;
 
         if ( type.compare( Header::BLOCK_CATCHUP_REQ ) == 0 ) {
-            serializedBinary = createBlockCatchupResponse( _jsonRequest,
+            serializedBinary = createBlockCatchupResponse( _connection, _jsonRequest,
                                                            dynamic_pointer_cast< CatchupResponseHeader >( _responseHeader ), blockID );
 
         } else if ( type.compare( Header::BLOCK_FINALIZE_REQ ) == 0 ) {
@@ -227,8 +227,8 @@ ptr< vector< uint8_t > > CatchupServerAgent::createResponseHeaderAndBinary(
 
 
 ptr< vector< uint8_t > > CatchupServerAgent::createBlockCatchupResponse(
-        nlohmann::json /*_jsonRequest */, const ptr< CatchupResponseHeader >& _responseHeader,
-        block_id _blockID ) {
+        const ptr< ServerConnection >& _connectionEnvelope, nlohmann::json /*_jsonRequest */,
+        const ptr< CatchupResponseHeader >& _responseHeader, block_id _blockID ) {
     CHECK_ARGUMENT( _responseHeader );
 
     MONITOR( __CLASS_NAME__, __FUNCTION__ );
@@ -279,7 +279,8 @@ ptr< vector< uint8_t > > CatchupServerAgent::createBlockCatchupResponse(
 
         auto responseTimeMs = Time::getCurrentTimeMs() - responseStartTimeMs;
 
-        LOG( info, "RETURNED_CATCHUP_BLOCKS:" << blockSizes->size() << ":CRT:" << responseTimeMs )
+        LOG( info, "RETURNED_CATCHUP_BLOCKS:" << blockSizes->size() << ":CRT:" << responseTimeMs
+             << ":TO_NODE:" << _connectionEnvelope->getIP() );
 
         return serializedBlocks;
     } catch ( ExitRequestedException& e ) {

catchup/server/CatchupServerAgent.h

@@ -44,8 +44,9 @@ class BlockFinalizeResponseHeader;
 class CatchupServerAgent : public AbstractServerAgent {
     ptr< CatchupWorkerThreadPool > catchupWorkerThreadPool;
 
-    ptr< vector< uint8_t > > createBlockCatchupResponse( nlohmann::json _jsonRequest,
-        const ptr< CatchupResponseHeader >& _responseHeader, block_id _blockID );
+    ptr< vector< uint8_t > > createBlockCatchupResponse( const ptr< ServerConnection >& _connectionEnvelope,
+        nlohmann::json _jsonRequest, const ptr< CatchupResponseHeader >& _responseHeader,
+        block_id _blockID );
 
 
     ptr< vector< uint8_t > > createBlockFinalizeResponse( nlohmann::json _jsonRequest,

chains/Schain.cpp

@@ -84,6 +84,7 @@
 #include "messages/NetworkMessageEnvelope.h"
 #include "monitoring/MonitoringAgent.h"
 #include "monitoring/StuckDetectionAgent.h"
+#include "monitoring/OptimizerAgent.h"
 #include "network/ClientSocket.h"
 #include "network/IO.h"
 #include "network/Sockets.h"
@@ -241,6 +242,7 @@ Schain::Schain( weak_ptr< Node > _node, schain_index _schainIndex, const schain_
       node( _node ),
       schainIndex( _schainIndex ) {
     lastCommittedBlockTimeStamp = TimeStamp( 0, 0 );
+    setTimeStampValuesFromConfig();
 
     // construct monitoring, timeout and stuck detection agents early
     monitoringAgent = make_shared< MonitoringAgent >( *this );
@@ -290,12 +292,6 @@ Schain::Schain( weak_ptr< Node > _node, schain_index _schainIndex, const schain_
 
         getNode()->registerAgent( this );
 
-
-        if ( getNode()->getPatchTimestamps().count( "verifyDaSigsPatchTimestamp" ) > 0 ) {
-            this->verifyDaSigsPatchTimestampS =
-                getNode()->getPatchTimestamps().at( "verifyDaSigsPatchTimestamp" );
-        }
-
     } catch ( ExitRequestedException& ) {
         throw;
     } catch ( ... ) {
@@ -309,6 +305,7 @@ void Schain::constructChildAgents() {
     MONITOR( __CLASS_NAME__, __FUNCTION__ )
 
     try {
+        optimizerAgent = make_shared< OptimizerAgent >( *this );
         oracleResultAssemblyAgent = make_shared< OracleResultAssemblyAgent >( *this );
         pricingAgent = make_shared< PricingAgent >( *this );
         catchupClientAgent = make_shared< CatchupClientAgent >( *this );
@@ -433,9 +430,12 @@ const atomic< bool >& Schain::getIsStateInitialized() const {
 }
 
 bool Schain::verifyDASigsPatch( uint64_t _blockTimeStampS ) {
-    return verifyDaSigsPatchTimestampS != 0 && _blockTimeStampS >= verifyDaSigsPatchTimestampS;
+    return verifyDaSigsPatchTimestamp != 0 && _blockTimeStampS >= verifyDaSigsPatchTimestamp;
 }
 
+bool Schain::verifyBlsSyncPatch( uint64_t _blockTimeStampS ) {
+    return verifyBlsSyncPatchTimestamp != 0 && _blockTimeStampS >= verifyBlsSyncPatchTimestamp;
+}
 
 void Schain::blockCommitArrived( block_id _committedBlockID, schain_index _proposerIndex,
     const ptr< ThresholdSignature >& _thresholdSig, ptr< ThresholdSignature > _daSig ) {
@@ -548,6 +548,13 @@ void Schain::proposeNextBlock( bool _isCalledAfterCatchup ) {
 
         proposedBlockArrived( myProposal );
 
+        if (getOptimizerAgent()->skipSendingProposalToTheNetwork(_proposedBlockID)) {
+            // a node skips sending and saving its proposal during
+            // optimized block consensus, if the node was not a winner
+            // last time
+            return; // dont propose
+        }
+
         LOG( debug, "PROPOSING BLOCK NUMBER:" << to_string( _proposedBlockID ) );
 
         auto db = getNode()->getProposalHashDB();
@@ -891,7 +898,10 @@ void Schain::daProofArrived( const ptr< DAProof >& _daProof ) {
         if ( _daProof->getBlockId() <= getLastCommittedBlockID() )
             return;
 
-        auto pv = getNode()->getDaProofDB()->addDAProof( _daProof );
+        // this will add the DAProof to DB. If there are enough DAProofs in DB
+        // to start binary consensus, this will return binary proposal vector of 1s and 0s
+        auto pv =
+            addDAProofToDBAndCalculateProposalVectorIfItsTimeToStartBinaryConsensus( _daProof );
 
 
         if ( pv != nullptr ) {
@@ -1461,11 +1471,62 @@ void Schain::analyzeErrors( ptr< CommittedBlock > _block ) {
         analyzer->analyze( _block );
     }
 }
-uint64_t Schain::getVerifyDaSigsPatchTimestampS() const {
-    return verifyDaSigsPatchTimestampS;
+uint64_t Schain::getVerifyDaSigsPatchTimeStamp() const {
+    return verifyDaSigsPatchTimestamp;
 }
 
-
+uint64_t Schain::getVerifyBlsSyncPatchTimestampS() const {
+    return verifyBlsSyncPatchTimestamp;
+}
 
 
 mutex Schain::vdsMutex;
+
+// this function is called on arrival of each DA proof
+// if it is time to make binary proposals it will return a vector of 0s and 1s
+// for normal consensus it will happen when 2t+1 DA proofs  arrive (which is 11)
+// for optimized consensus it will happen when a DA proof from the previous winner arrives
+ptr<BooleanProposalVector>
+Schain::addDAProofToDBAndCalculateProposalVectorIfItsTimeToStartBinaryConsensus(
+    const ptr<DAProof> &_daProof) {
+
+    ptr<BooleanProposalVector> pv;
+
+    if (getOptimizerAgent()->doOptimizedConsensus(_daProof->getBlockId(), getLastCommittedBlockTimeStamp().getS())) {
+        // when we do optimized block consensus only the previous winner
+        // proposes and provides da proof
+        // proposals from other nodes, if sent made by mistake, are ignored
+        auto lastWinner = getOptimizerAgent()->getPreviousWinner( _daProof->getBlockId() );
+        if (_daProof->getProposerIndex() == lastWinner) {
+            getNode()->getDaProofDB()->addDAProof(_daProof);
+            pv = make_shared<BooleanProposalVector>(getNodeCount(), lastWinner);
+        }
+    } else {
+        // do things regular way
+        // the binary proposal vector is formed and the consensus is started when
+        // 2/3 of nodes  (11) submit a da proof
+        pv = getNode()->getDaProofDB()->addDAProof(_daProof);
+    }
+    return pv;
+}
+
+// returns true if fastConsensusPatch ie enabled
+bool Schain::fastConsensusPatchEnabled(uint64_t _blockTimeStampSec ) {
+    return fastConsensusPatchTimestamp != 0 && _blockTimeStampSec >= fastConsensusPatchTimestamp;
+}
+
+// macro to set patchstamp variable from connfig
+#define SET_TIMESTAMP_FROM_CONFIG(__TIMESTAMP_NAME__) \
+    { \
+        auto& timestamps = getNode()->getPatchTimestamps(); \
+        if (timestamps.count(#__TIMESTAMP_NAME__) > 0) { \
+            __TIMESTAMP_NAME__ = timestamps.at(#__TIMESTAMP_NAME__); \
+        } \
+    }
+
+// set all timestamp values from config
+void Schain::setTimeStampValuesFromConfig() {
+    SET_TIMESTAMP_FROM_CONFIG(verifyDaSigsPatchTimestamp)
+    SET_TIMESTAMP_FROM_CONFIG(fastConsensusPatchTimestamp)
+    SET_TIMESTAMP_FROM_CONFIG(verifyBlsSyncPatchTimestamp)
+}

chains/Schain.h

@@ -61,7 +61,7 @@ class CatchupServerAgent;
 class MonitoringAgent;
 class TimeoutAgent;
 class StuckDetectionAgent;
-
+class OptimizerAgent;
 
 class BlockProposalServerAgent;
 
@@ -136,6 +136,7 @@ class Schain : public Agent {
 
     ptr< OracleResultAssemblyAgent > oracleResultAssemblyAgent;
 
+    ptr<OptimizerAgent> optimizerAgent;
 
     ptr< IO > io;
 
@@ -175,7 +176,9 @@ class Schain : public Agent {
 
     ptr< NodeInfo > thisNodeInfo = nullptr;
 
-    uint64_t verifyDaSigsPatchTimestampS = 0;
+    uint64_t verifyDaSigsPatchTimestamp = 0;
+    uint64_t fastConsensusPatchTimestamp = 0;
+    uint64_t verifyBlsSyncPatchTimestamp = 0;
 
     // If a BlockError analyzer is added to the queue
     // its analyze(CommittedBlock _block) function will be run on commit
@@ -344,9 +347,9 @@ class Schain : public Agent {
 
     ptr< CryptoManager > getCryptoManager() const;
 
+    uint64_t getVerifyDaSigsPatchTimeStamp() const;
 
-    uint64_t getVerifyDaSigsPatchTimestampS() const;
-
+    uint64_t getVerifyBlsSyncPatchTimestampS() const;
 
     bool isInCreateBlock() const;
 
@@ -386,8 +389,18 @@ class Schain : public Agent {
 
     bool verifyDASigsPatch( uint64_t _blockTimeStampSec );
 
+    bool verifyBlsSyncPatch( uint64_t _blockTimeStampSec );
+
     void updateInternalChainInfo( block_id _lastCommittedBlockID );
 
     const ptr<CatchupClientAgent> &getCatchupClientAgent() const;
 
+    ptr< OptimizerAgent > getOptimizerAgent() const;
+
+    bool fastConsensusPatchEnabled( uint64_t _blockTimeStampSec );
+
+    void setTimeStampValuesFromConfig();
+
+    ptr<BooleanProposalVector>
+    addDAProofToDBAndCalculateProposalVectorIfItsTimeToStartBinaryConsensus(const ptr<DAProof> &_daProof);
 };

chains/SchainGettersSetters.cpp

@@ -268,6 +268,13 @@ ptr< CryptoManager > Schain::getCryptoManager() const {
     return cryptoManager;
 }
 
+
+ptr< OptimizerAgent > Schain::getOptimizerAgent() const {
+    CHECK_STATE( optimizerAgent );
+    return optimizerAgent;
+}
+
+
 void Schain::createBlockConsensusInstance() {
     blockConsensusInstance = make_shared< BlockConsensusAgent >( *this );
 }
@@ -320,8 +327,6 @@ void Schain::updateLastCommittedBlockInfo( uint64_t _lastCommittedBlockID,
     tpsAverage = ( blockSizeAverage * 1000 ) / blockTimeAverageMs;
     getRandomForBlockId( ( uint64_t ) lastCommittedBlockID );
 
-    if ( getNode()->isSyncOnlyNode() )
-        return;
 }
 
 

crypto/CryptoManager.cpp

@@ -754,10 +754,14 @@ void CryptoManager::verifyBlockSig(
         }
 
         if ( verifyRealSignatures ) {
-            auto _signature = make_shared< ConsensusBLSSignature >(
-                _sigStr, _blockId, totalSigners, requiredSigners );
+            if ( !getSchain()->getNode()->isSyncOnlyNode() ||
+                 ( getSchain()->getNode()->isSyncOnlyNode() &&
+                   getSchain()->verifyBlsSyncPatch( _ts.getS() ) ) ) {
+                auto _signature = make_shared< ConsensusBLSSignature >(
+                    _sigStr, _blockId, totalSigners, requiredSigners );
 
-            verifyThresholdSig( _signature, _hash, _ts );
+                verifyThresholdSig( _signature, _hash, _ts );
+            }
         }
 
     } catch ( ... ) {
@@ -880,7 +884,8 @@ void CryptoManager::verifyThresholdSig(
 
         MONITOR( __CLASS_NAME__, __FUNCTION__ )
 
-        if ( verifyRealSignatures ) {
+        if ( verifyRealSignatures && ( !getSchain()->getNode()->isSyncOnlyNode() ||
+                                       getSchain()->verifyBlsSyncPatch( _ts.getS() ) ) ) {
             auto blsSig = dynamic_pointer_cast< ConsensusBLSSignature >( _signature );
 
             CHECK_STATE( blsSig );
@@ -907,7 +912,6 @@ void CryptoManager::verifyThresholdSig(
                         make_shared< array< uint8_t, HASH_LEN > >( _hash.getHash() ), libBlsSig ),
                     "BLS sig verification failed using both current and previous key" );
             }
-
         } else {
             // mockups sigs are not verified
         }

datastructures/BooleanProposalVector.cpp

@@ -63,6 +63,27 @@ BooleanProposalVector::BooleanProposalVector( node_count _nodeCount, const strin
 }
 
 
+
+// This is boolean proposal constructor in case of optimized proposal. In this
+// case we proposer 1 for the previos winner and 0 for everyone else
+BooleanProposalVector::BooleanProposalVector( node_count _nodeCount, schain_index _previousWinner )
+        : nodeCount( _nodeCount ) {
+
+    CHECK_ARGUMENT( _previousWinner > 0 );
+    CHECK_ARGUMENT( _previousWinner <= (uint64_t )_nodeCount );
+    proposals.push_back( false );
+
+    for ( uint64_t i = 1; i <= _nodeCount; i++ ) {
+        if ( i == _previousWinner ) {
+            proposals.push_back( true );
+            trueCount++;
+        } else {
+            proposals.push_back( false );
+        }
+    }
+}
+
+
 bool BooleanProposalVector::getProposalValue( schain_index _index ) {
     CHECK_STATE( proposals.size() == nodeCount + 1 );
     CHECK_STATE( _index <= ( uint64_t ) nodeCount );