Fix utxopool udpate bug for invalid cross shard tx (finally the consensus is stable with 30% cross shard tx); reset consensus before retry

aws-code/txgen/main.go

@@ -219,7 +219,7 @@ func main() {
 	config := readConfigFile(*configFile)
 	leaders, shardIds := getLeadersAndShardIds(&config)
 
-	crossShard := false //len(shardIds) > 1
+	crossShard := len(shardIds) > 1
 
 	// Setup a logger to stdout and log file.
 	logFileName := fmt.Sprintf("./%v/tx-generator.log", *logFolder)

blockchain/utxopool.go

@@ -155,6 +155,21 @@ func (utxoPool *UTXOPool) UpdateOneTransaction(tx *Transaction) {
 		}
 	}
 
+	isValidCrossShard := true
+	if isCrossShard {
+		// Check whether for this shard this cross transaction is valid or not.
+		for _, in := range tx.TxInput {
+			// Only check the input for my own shard.
+			if in.ShardId != utxoPool.ShardId {
+				continue
+			}
+			inTxID := hex.EncodeToString(in.TxID[:])
+			if _, ok := utxoPool.UtxoMap[in.Address][inTxID][in.TxOutputIndex]; !ok {
+				isValidCrossShard = false
+			}
+		}
+	}
+
 	utxoPool.mutex.Lock()
 	defer utxoPool.mutex.Unlock()
 	if utxoPool != nil {
@@ -162,27 +177,29 @@ func (utxoPool *UTXOPool) UpdateOneTransaction(tx *Transaction) {
 
 		// Remove
 		if !isUnlockTx {
-			for _, in := range tx.TxInput {
-				// Only check the input for my own shard.
-				if in.ShardId != utxoPool.ShardId {
-					continue
-				}
+			if isValidCrossShard {
+				for _, in := range tx.TxInput {
+					// Only check the input for my own shard.
+					if in.ShardId != utxoPool.ShardId {
+						continue
+					}
 
-				// NOTE: for the locking phase of cross tx, the utxo is simply removed from the pool.
-				inTxID := hex.EncodeToString(in.TxID[:])
-				value := utxoPool.UtxoMap[in.Address][inTxID][in.TxOutputIndex]
-				utxoPool.DeleteOneUtxo(in.Address, inTxID, in.TxOutputIndex)
-				if isCrossShard {
-					// put the delete (locked) utxo into a separate locked utxo pool
+					// NOTE: for the locking phase of cross tx, the utxo is simply removed from the pool.
 					inTxID := hex.EncodeToString(in.TxID[:])
-					if _, ok := utxoPool.LockedUtxoMap[in.Address]; !ok {
-						utxoPool.LockedUtxoMap[in.Address] = make(map[string]map[int]int)
-						utxoPool.LockedUtxoMap[in.Address][inTxID] = make(map[int]int)
+					value := utxoPool.UtxoMap[in.Address][inTxID][in.TxOutputIndex]
+					utxoPool.DeleteOneUtxo(in.Address, inTxID, in.TxOutputIndex)
+					if isCrossShard {
+						// put the delete (locked) utxo into a separate locked utxo pool
+						inTxID := hex.EncodeToString(in.TxID[:])
+						if _, ok := utxoPool.LockedUtxoMap[in.Address]; !ok {
+							utxoPool.LockedUtxoMap[in.Address] = make(map[string]map[int]int)
+							utxoPool.LockedUtxoMap[in.Address][inTxID] = make(map[int]int)
+						}
+						if _, ok := utxoPool.LockedUtxoMap[in.Address][inTxID]; !ok {
+							utxoPool.LockedUtxoMap[in.Address][inTxID] = make(map[int]int)
+						}
+						utxoPool.LockedUtxoMap[in.Address][inTxID][in.TxOutputIndex] = value
 					}
-					if _, ok := utxoPool.LockedUtxoMap[in.Address][inTxID]; !ok {
-						utxoPool.LockedUtxoMap[in.Address][inTxID] = make(map[int]int)
-					}
-					utxoPool.LockedUtxoMap[in.Address][inTxID][in.TxOutputIndex] = value
 				}
 			}
 		}
@@ -190,26 +207,28 @@ func (utxoPool *UTXOPool) UpdateOneTransaction(tx *Transaction) {
 		// Update
 		if !isCrossShard || isUnlockTx {
 			if !unlockToCommit {
-				// unlock-to-abort, bring back (unlock) the utxo input
-				for _, in := range tx.TxInput {
-					// Only unlock the input for my own shard.
-					if in.ShardId != utxoPool.ShardId {
-						continue
+				if isValidCrossShard {
+					// unlock-to-abort, bring back (unlock) the utxo input
+					for _, in := range tx.TxInput {
+						// Only unlock the input for my own shard.
+						if in.ShardId != utxoPool.ShardId {
+							continue
+						}
+
+						// Simply bring back the locked (removed) utxo
+						inTxID := hex.EncodeToString(in.TxID[:])
+						if _, ok := utxoPool.UtxoMap[in.Address]; !ok {
+							utxoPool.UtxoMap[in.Address] = make(map[string]map[int]int)
+							utxoPool.UtxoMap[in.Address][inTxID] = make(map[int]int)
+						}
+						if _, ok := utxoPool.UtxoMap[in.Address][inTxID]; !ok {
+							utxoPool.UtxoMap[in.Address][inTxID] = make(map[int]int)
+						}
+						value := utxoPool.LockedUtxoMap[in.Address][inTxID][in.TxOutputIndex]
+						utxoPool.UtxoMap[in.Address][inTxID][in.TxOutputIndex] = value
+
+						utxoPool.DeleteOneLockedUtxo(in.Address, inTxID, in.TxOutputIndex)
 					}
-
-					// Simply bring back the locked (removed) utxo
-					inTxID := hex.EncodeToString(in.TxID[:])
-					if _, ok := utxoPool.UtxoMap[in.Address]; !ok {
-						utxoPool.UtxoMap[in.Address] = make(map[string]map[int]int)
-						utxoPool.UtxoMap[in.Address][inTxID] = make(map[int]int)
-					}
-					if _, ok := utxoPool.UtxoMap[in.Address][inTxID]; !ok {
-						utxoPool.UtxoMap[in.Address][inTxID] = make(map[int]int)
-					}
-					value := utxoPool.LockedUtxoMap[in.Address][inTxID][in.TxOutputIndex]
-					utxoPool.UtxoMap[in.Address][inTxID][in.TxOutputIndex] = value
-
-					utxoPool.DeleteOneLockedUtxo(in.Address, inTxID, in.TxOutputIndex)
 				}
 			} else {
 				// normal utxo output update
@@ -237,16 +256,9 @@ func (utxoPool *UTXOPool) UpdateOneTransaction(tx *Transaction) {
 // VerifyOneTransactionAndUpdate verifies and update a valid transaction.
 // Return false if the transaction is not valid.
 func (utxoPool *UTXOPool) VerifyOneTransactionAndUpdate(tx *Transaction) bool {
-	if valid, crossShard := utxoPool.VerifyOneTransaction(tx, nil); valid {
+	if valid, _ := utxoPool.VerifyOneTransaction(tx, nil); valid {
 		utxoPool.UpdateOneTransaction(tx)
-		if crossShard {
-			// TODO: send proof-of-accceptance
-		}
 		return true
-	} else if crossShard {
-		if crossShard {
-			// TODO: send proof-of-rejection
-		}
 	}
 	return false
 }

consensus/consensus.go

@@ -19,8 +19,6 @@ type Consensus struct {
 	commits map[string]string
 	// Signatures collected from validators
 	responses map[string]string
-	// Actual block data to reach consensus on
-	data string
 	// List of validators
 	validators []p2p.Peer
 	// Leader

node/node_handler.go

@@ -190,6 +190,7 @@ func (node *Node) WaitForConsensusReady(readySignal chan int) {
 			time.Sleep(100 * time.Millisecond) // Delay a bit so validator is catched up.
 		case <-time.After(8 * time.Second):
 			retry = true
+			node.Consensus.ResetState()
 			timeoutCount++
 			node.log.Debug("Consensus timeout, retry!", "count", timeoutCount, "node", node)
 		}