derohe-miniblock-mod/p2p/chain_sync.go

// Copyright 2017-2021 DERO Project. All rights reserved.
// Use of this source code in any form is governed by RESEARCH license.
// license can be found in the LICENSE file.
// GPG: 0F39 E425 8C65 3947 702A  8234 08B2 0360 A03A 9DE8
//
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

package p2p

import "fmt"
import "time"
import "math/big"
import "sync/atomic"

import "github.com/deroproject/derohe/config"
import "github.com/deroproject/derohe/globals"
import "github.com/deroproject/derohe/block"
import "github.com/deroproject/derohe/errormsg"
import "github.com/deroproject/derohe/blockchain"
import "github.com/deroproject/derohe/transaction"
import "github.com/deroproject/derohe/cryptography/crypto"

// used to satisfy difficulty interface
type MemorySource struct {
	Blocks     map[crypto.Hash]*block.Complete_Block
	Difficulty map[crypto.Hash]*big.Int
}

// local blocks are inserted using this
func (x *MemorySource) insert_block_noprecheck(cbl *block.Complete_Block, diff *big.Int) {
	blid := cbl.Bl.GetHash()
	x.Blocks[blid] = cbl
	x.Difficulty[blid] = diff
}

// remote peers blocks are added using this
func (x *MemorySource) insert_block(cbl *block.Complete_Block) {
	if len(cbl.Bl.Tips) == 0 {
		panic("genesis block not possible")
	}
	diff := blockchain.Get_Difficulty_At_Tips(x, cbl.Bl.Tips)
	blid := cbl.Bl.GetHash()

	for _, mbl := range cbl.Bl.MiniBlocks {
		PoW := mbl.GetPoWHash()
		if blockchain.CheckPowHashBig(PoW, diff) == false {
			panic("pow failed")
		}
	}

	x.Blocks[blid] = cbl
	x.Difficulty[blid] = diff
}
func (x *MemorySource) check(blid crypto.Hash) *block.Complete_Block {
	cbl, ok := x.Blocks[blid]
	if !ok {
		panic(fmt.Errorf("no such blid in ram %s", blid))
	}
	return cbl
}
func (x *MemorySource) Load_Block_Height(blid crypto.Hash) int64 {
	return int64(x.check(blid).Bl.Height)
}
func (x *MemorySource) Load_Block_Timestamp(blid crypto.Hash) uint64 {
	return uint64(x.check(blid).Bl.Timestamp)
}
func (x *MemorySource) Get_Block_Past(blid crypto.Hash) []crypto.Hash {
	return x.check(blid).Bl.Tips
}
func (x *MemorySource) Load_Block_Difficulty(blid crypto.Hash) *big.Int {
	diff, ok := x.Difficulty[blid]
	if !ok {
		panic(fmt.Errorf("no such blid in ram %s", blid))
	}
	return diff
}

// convert block which was serialized in p2p format to in ram block format
func ConvertCBlock_To_CompleteBlock(cblock Complete_Block) (cbl block.Complete_Block, diff *big.Int) {
	var bl block.Block
	cbl.Bl = &bl
	if err := bl.Deserialize(cblock.Block); err != nil {
		panic(err)
	}
	// complete the txs
	for j := range cblock.Txs {
		var tx transaction.Transaction
		if err := tx.Deserialize(cblock.Txs[j]); err != nil { // we have a tx which could not be deserialized ban peer
			panic(err)
		}
		cbl.Txs = append(cbl.Txs, &tx)
	}

	if len(bl.Tx_hashes) != len(cbl.Txs) {
		panic(fmt.Errorf("txcount mismatch, expected %d txs actual %d", len(bl.Tx_hashes), len(cbl.Txs)))
	}

	return
}

// we are expecting other side to have a heavier PoW chain, try to sync now
func (connection *Connection) sync_chain() {

	defer handle_connection_panic(connection)

	var request Chain_Request_Struct
	var response Chain_Response_Struct

try_again:

	// send our blocks, first 10 blocks directly, then decreasing in powers of 2
	start_point := chain.Load_TOPO_HEIGHT()
	for i := int64(0); i < start_point; {

		tr, err := chain.Store.Topo_store.Read(start_point - i) // commit everything
		if err != nil {
			continue
		}
		if tr.IsClean() {
			break
		}

		request.Block_list = append(request.Block_list, tr.BLOCK_ID)
		request.TopoHeights = append(request.TopoHeights, start_point-i)
		switch {
		case len(request.Block_list) < 20: // 20 blocks raw
			i++
		case len(request.Block_list) < 100: // 20 block with 5 steps
			i += 5
		case len(request.Block_list) < 1000: // 20 block with 50 steps
			i += 50
		case len(request.Block_list) < 10000: // 20 block with 500 steps
			i += 500
		default:
			i = i * 2
		}
	}
	// add genesis block at the end
	request.Block_list = append(request.Block_list, globals.Config.Genesis_Block_Hash)
	request.TopoHeights = append(request.TopoHeights, 0)
	fill_common(&request.Common) // fill common info

	if err := connection.Client.Call("Peer.Chain", request, &response); err != nil {
		connection.logger.V(2).Error(err, "Call failed Chain")
		return
	}
	// we have a response, see if its valid and try to add to get the blocks

	connection.logger.V(2).Info("Peer wants to give chain", "from topoheight", response.Start_height)

	var pop_count int64

	// we do not need reorganisation if deviation is less than  or equak to 7 blocks
	// only pop blocks if the system has somehow deviated more than 7 blocks
	// if the deviation is less than 7 blocks, we internally reorganise everything
	if chain.Get_Height()-response.Start_height > config.STABLE_LIMIT && connection.SyncNode {
		// get our top block
		connection.logger.V(3).Info("rewinding status", "our topoheight", chain.Load_TOPO_HEIGHT(), "peer topoheight", response.Start_topoheight)
		pop_count := chain.Load_TOPO_HEIGHT() - response.Start_topoheight
		chain.Rewind_Chain(int(pop_count)) // pop as many blocks as necessary

		// we should NOT queue blocks, instead we sent our chain request again
		goto try_again

	} else if chain.Get_Height()-response.Common.Height != 0 && chain.Get_Height()-response.Start_height <= config.STABLE_LIMIT {
		pop_count = chain.Load_TOPO_HEIGHT() - response.Start_topoheight
	} else if chain.Get_Height() < connection.Height && chain.Get_Height()-response.Start_height > config.STABLE_LIMIT { // we must somehow notify that deviation is way too much and manual interaction is necessary, so as any bug for chain deviationmay be detected
		//connection.logger.V(1).Error(nil, "we have or others have deviated too much.you may have to use --sync-node option", "our topoheight", chain.Load_TOPO_HEIGHT(), "peer topoheight start", response.Start_topoheight)
		//return
		pop_count = chain.Load_TOPO_HEIGHT() - response.Start_topoheight
	}

	if pop_count >= 1 { // peer is claiming his chain is good and we should rewind
		connection.logger.V(1).Info("syncing", "pop_count", pop_count)

		ramstore := MemorySource{Blocks: map[crypto.Hash]*block.Complete_Block{}, Difficulty: map[crypto.Hash]*big.Int{}}
		start_point := response.Start_topoheight
		for i := 0; i < 10 && start_point >= 1; i++ {
			start_point--
		}

		for i := start_point; i < start_point+128 && i <= chain.Load_TOPO_HEIGHT(); i++ {
			blid, err := chain.Load_Block_Topological_order_at_index(i)
			if err != nil {
				connection.logger.V(1).Info("error loading local topo", "i", i)
				return
			}
			cbl, err := chain.Load_Complete_Block(blid)
			if err != nil {
				connection.logger.V(1).Info("error loading completeblock", "blid", blid)
				return
			}
			diff := chain.Load_Block_Difficulty(blid)
			ramstore.insert_block_noprecheck(cbl, diff) // all local blocks are already checked
		}

		// now we must request peer blocks and verify their pow

		for i := range response.Block_list {
			our_topo_order := chain.Load_Block_Topological_order(response.Block_list[i])
			if our_topo_order != (int64(i)+response.Start_topoheight) || our_topo_order == -1 { // if block is not in our chain, add it to request list

				var orequest ObjectList
				var oresponse Objects

				//fmt.Printf("inserting blocks %d %x\n", (int64(i) + response.Start_topoheight), response.Block_list[i][:])
				orequest.Block_list = append(orequest.Block_list, response.Block_list[i])
				fill_common(&orequest.Common)
				if err := connection.Client.Call("Peer.GetObject", orequest, &oresponse); err != nil {
					connection.logger.V(2).Error(err, "Call failed GetObject")
					return
				} else { // process the response
					cbl, _ := ConvertCBlock_To_CompleteBlock(oresponse.CBlocks[0])
					ramstore.insert_block(&cbl) // insert block with checking verification
				}
			}
		}

		// if we reached here we were able to verify basic chain structure

		//	chain.Rewind_Chain(int(pop_count)) // pop as many blocks as necessary, assumming peer has given us good chain

		failcount := 0
		for i := range response.Block_list {
			our_topo_order := chain.Load_Block_Topological_order(response.Block_list[i])
			if our_topo_order != (int64(i)+response.Start_topoheight) || our_topo_order == -1 { // if block is not in our chain, add it to request list
				if failcount < 4 {
					if _, ok := chain.Add_Complete_Block(ramstore.check(response.Block_list[i])); !ok {
						failcount++
					}
				}
			}
		}

		connection.logger.V(1).Info("rewinded blocks", "pop_count", pop_count)
		return
	}

	// response only 128 blocks at a time
	max_blocks_to_queue := 128
	// check whether the objects are in our db or not
	// until we put in place a parallel object tracker, do it one at a time

	connection.logger.V(2).Info("response block list", "count", len(response.Block_list))
	for i := range response.Block_list {
		our_topo_order := chain.Load_Block_Topological_order(response.Block_list[i])
		if our_topo_order != (int64(i)+response.Start_topoheight) || our_topo_order == -1 { // if block is not in our chain, add it to request list
			//queue_block(request.Block_list[i])
			if max_blocks_to_queue >= 0 {
				max_blocks_to_queue--
				//connection.Send_ObjectRequest([]crypto.Hash{response.Block_list[i]}, []crypto.Hash{})
				var orequest ObjectList
				var oresponse Objects

				orequest.Block_list = append(orequest.Block_list, response.Block_list[i])
				fill_common(&orequest.Common)
				if err := connection.Client.Call("Peer.GetObject", orequest, &oresponse); err != nil {
					connection.logger.V(2).Error(err, "Call failed GetObject")
					return
				} else { // process the response
					if err = connection.process_object_response(oresponse, 0, true); err != nil {
						return
					}
				}

				//	fmt.Printf("Queuing block %x height %d  %s", response.Block_list[i], response.Start_height+int64(i), connection.logid)
			}
		} else {
			connection.logger.V(3).Info("We must have queued but we skipped it at height", "blid", fmt.Sprintf("%x", response.Block_list[i]), "height", response.Start_height+int64(i))
		}
	}

	// request alt-tips ( blocks if we are nearing the main tip )
	/*if (response.Common.TopoHeight - chain.Load_TOPO_HEIGHT()) <= 5 {
		for i := range response.TopBlocks {
			if chain.Load_Block_Topological_order(response.TopBlocks[i]) == -1 {
				//connection.Send_ObjectRequest([]crypto.Hash{response.TopBlocks[i]}, []crypto.Hash{})
				connection.logger.V(2).Info("Queuing ALT-TIP", "blid", response.TopBlocks[i])

			}

		}
	}*/

}

func (connection *Connection) process_object_response(response Objects, sent int64, syncing bool) error {
	var err error
	defer globals.Recover(2)

	for i := 0; i < len(response.CBlocks); i++ { // process incoming full blocks
		// make sure connection does not timeout and be killed while processing huge blocks

		atomic.StoreInt64(&connection.LastObjectRequestTime, time.Now().Unix())

		var cbl block.Complete_Block // parse incoming block and deserialize it
		var bl block.Block
		// lets deserialize block first and see whether it is the requested object
		cbl.Bl = &bl
		err := bl.Deserialize(response.CBlocks[i].Block)
		if err != nil { // we have a block which could not be deserialized ban peer
			connection.logger.V(2).Error(err, "Incoming block could not be deserilised")
			connection.exit()
			if syncing {
				return nil
			} else {
				return err
			}
		}

		// give the chain some more time to respond
		atomic.StoreInt64(&connection.LastObjectRequestTime, time.Now().Unix())

		// do not try to add blocks too much into future
		if syncing && int64(bl.Height) > chain.Get_Height()+4 {
			continue
		}
		// check whether the object was requested one

		// complete the txs
		for j := range response.CBlocks[i].Txs {
			var tx transaction.Transaction
			err = tx.Deserialize(response.CBlocks[i].Txs[j])
			if err != nil { // we have a tx which could not be deserialized ban peer
				connection.logger.V(2).Error(err, "Incoming TX could not be deserilised")
				connection.exit()

				if syncing {
					return nil
				} else {
					return err
				}
			}
			cbl.Txs = append(cbl.Txs, &tx)
		}

		// check if we can add ourselves to chain
		err, ok := chain.Add_Complete_Block(&cbl)
		if !ok && err == errormsg.ErrInvalidPoW {
			connection.logger.V(2).Error(err, "This peer should be banned")
			connection.exit()
			if syncing {
				return nil
			} else {
				return err
			}
		}

		if !ok && err == errormsg.ErrPastMissing {
			connection.logger.V(2).Error(err, "Incoming Block could not be added due to missing past, so skipping future block")
			if syncing {
				return nil
			} else {
				return err
			}
		}

		if !ok {
			connection.logger.V(2).Error(err, "Incoming Block could not be added due to some error")
		}

		// add the object to object pool from where it will be consume
		// queue_block_received(bl.GetHash(),&cbl)

	}

	for i := range response.Txs { // process incoming txs for mempool
		var tx transaction.Transaction
		err = tx.Deserialize(response.Txs[i])
		if err != nil { // we have a tx which could not be deserialized ban peer
			connection.logger.V(2).Error(err, "Incoming TX could not be deserilised")
			connection.exit()

			return nil
		}

		if !chain.Mempool.Mempool_TX_Exist(tx.GetHash()) { // we still donot have it, so try to process it
			if chain.Add_TX_To_Pool(&tx) == nil { // currently we are ignoring error
				broadcast_Tx(&tx, 0, sent)
			}
		}
	}

	for i := range response.Chunks { // process incoming chunks
		connection.feed_chunk(&response.Chunks[i], sent)
	}

	return nil
}