feat(node): reduce blocks request time (#185)

.github/workflows/quality.yml

@@ -15,6 +15,17 @@ jobs:
         with:
           go-version: 1.18
       - uses: golangci/golangci-lint-action@v3
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v3
+      - name: Setup
+        uses: actions/setup-go@v3
+        with:
+          go-version: 1.17
+      - name: Test
+        run: go test -v ./...
   scan:
     runs-on: ubuntu-latest
     steps:
@@ -25,7 +36,7 @@ jobs:
       with:
         go-version: 1.17
     - name: Test
-      run: go test -v -coverpkg=./... ./... -coverprofile=coverage.out -json > report.json
+      run: go test -coverpkg=./... ./... -coverprofile=coverage.out -json > report.json
     - name: SonarCloud Scan
       uses: SonarSource/sonarcloud-github-action@master
       env:

src/node/main.go

@@ -54,10 +54,13 @@ func main() {
 	}
 	synchronizationTimer := time.Second * synchronizationIntervalInSeconds
 	synchronizationEngine := tick.NewEngine(synchronizer.Synchronize, watch, synchronizationTimer, 1, 0)
-	blockchain := verification.NewBlockchain(registry, validationTimer, synchronizer, logger)
-	pool := validation.NewTransactionsPool(blockchain, registry, wallet.Address(), settings.GenesisAmount, validationTimer, watch, logger)
+	now := watch.Now()
+	initialTimestamp := now.Truncate(validationTimer).Add(validationTimer).UnixNano()
+	genesisTransaction := validation.NewRewardTransaction(wallet.Address(), initialTimestamp, settings.GenesisAmount)
+	blockchain := verification.NewBlockchain(genesisTransaction, registry, validationTimer, synchronizer, logger)
+	pool := validation.NewTransactionsPool(blockchain, registry, wallet.Address(), validationTimer, watch, logger)
 	validationEngine := tick.NewEngine(pool.Validate, watch, validationTimer, 1, 0)
-	verificationEngine := tick.NewEngine(blockchain.Verify, watch, validationTimer, verificationsCountPerValidation, 1)
+	verificationEngine := tick.NewEngine(blockchain.Update, watch, validationTimer, verificationsCountPerValidation, 1)
 	serverFactory := gp2p.NewServerFactory()
 	server, err := serverFactory.CreateServer(int(*port))
 	if err != nil {

src/node/network/last_bocks_request.go

@@ -0,0 +1,5 @@
+package network
+
+type LastBlocksRequest struct {
+	StartingBlockHash *[32]byte
+}

src/node/network/neighbor.go

@@ -5,6 +5,7 @@ type Neighbor interface {
 	Port() uint16
 	Target() string
 	GetBlocks() (blockResponses []*BlockResponse, err error)
+	GetLastBlocks(lastBlocksRequest LastBlocksRequest) ([]*BlockResponse, error)
 	SendTargets(request []TargetRequest) (err error)
 	AddTransaction(request TransactionRequest) (err error)
 	GetTransactions() (transactionResponses []TransactionResponse, err error)

src/node/network/p2p/client.go

@@ -4,4 +4,5 @@ import gp2p "github.com/leprosus/golang-p2p"
 
 type Client interface {
 	Send(topic string, req gp2p.Data) (res gp2p.Data, err error)
+	SetSettings(stg *gp2p.ClientSettings)
 }

src/node/network/p2p/gp2p/client_factory.go

@@ -8,11 +8,8 @@ import (
 	"github.com/my-cloud/ruthenium/src/node/network"
 	"github.com/my-cloud/ruthenium/src/node/network/p2p"
 	"strconv"
-	"time"
 )
 
-const clientConnectionTimeoutInSeconds = 10
-
 type ClientFactory struct {
 	ipFinder network.IpFinder
 	logger   log.Logger
@@ -37,10 +34,6 @@ func (factory *ClientFactory) CreateClient(ip string, port uint16) (p2p.Client,
 	if err != nil {
 		return nil, err
 	}
-	settings := gp2p.NewClientSettings()
-	settings.SetRetry(1, time.Nanosecond)
-	settings.SetConnTimeout(clientConnectionTimeoutInSeconds * time.Second)
-	client.SetSettings(settings)
 	client.SetLogger(factory.logger)
 	return client, err
 }

src/node/network/p2p/host.go

@@ -65,6 +65,7 @@ func (host *Host) startBlockchain() {
 func (host *Host) handle(_ context.Context, req gp2p.Data) (res gp2p.Data, err error) {
 	var unknownRequest bool
 	var requestString string
+	var lastBlocksRequest network.LastBlocksRequest
 	var transactionRequest network.TransactionRequest
 	var amountRequest network.AmountRequest
 	var targetsRequest []network.TargetRequest
@@ -82,6 +83,8 @@ func (host *Host) handle(_ context.Context, req gp2p.Data) (res gp2p.Data, err e
 		default:
 			unknownRequest = true
 		}
+	} else if err = req.GetGob(&lastBlocksRequest); err == nil {
+		res = host.getLastBlocks(&lastBlocksRequest)
 	} else if err = req.GetGob(&transactionRequest); err == nil {
 		host.addTransactions(&transactionRequest)
 	} else if err = req.GetGob(&amountRequest); err == nil {
@@ -98,6 +101,15 @@ func (host *Host) handle(_ context.Context, req gp2p.Data) (res gp2p.Data, err e
 	return
 }
 
+func (host *Host) getLastBlocks(request *network.LastBlocksRequest) (res gp2p.Data) {
+	blockResponses := host.blockchain.LastBlocks(*request.StartingBlockHash)
+	err := res.SetGob(blockResponses)
+	if err != nil {
+		host.logger.Error(fmt.Errorf("failed to get blocks: %w", err).Error())
+	}
+	return
+}
+
 func (host *Host) getBlocks() (res gp2p.Data) {
 	blockResponses := host.blockchain.Blocks()
 	err := res.SetGob(blockResponses)

src/node/network/p2p/neighbor.go

@@ -4,19 +4,30 @@ import (
 	"fmt"
 	gp2p "github.com/leprosus/golang-p2p"
 	"github.com/my-cloud/ruthenium/src/node/network"
+	"time"
+)
+
+const (
+	initializationConnectionTimeoutInSeconds = 600
+	commonConnectionTimeoutInSeconds         = 5
 )
 
 type Neighbor struct {
-	target *Target
-	client Client
+	target   *Target
+	client   Client
+	settings *gp2p.ClientSettings
 }
 
 func NewNeighbor(target *Target, clientFactory ClientFactory) (*Neighbor, error) {
 	client, err := clientFactory.CreateClient(target.Ip(), target.Port())
 	if err != nil {
 		return nil, fmt.Errorf("failed to create client reaching %s: %w", target.Value(), err)
 	}
-	return &Neighbor{target, client}, nil
+	settings := gp2p.NewClientSettings()
+	settings.SetRetry(1, time.Nanosecond)
+	settings.SetConnTimeout(commonConnectionTimeoutInSeconds * time.Second)
+	client.SetSettings(settings)
+	return &Neighbor{target, client, settings}, nil
 }
 
 func (neighbor *Neighbor) Ip() string {
@@ -32,10 +43,22 @@ func (neighbor *Neighbor) Target() string {
 }
 
 func (neighbor *Neighbor) GetBlocks() (blockResponses []*network.BlockResponse, err error) {
+	neighbor.settings.SetConnTimeout(initializationConnectionTimeoutInSeconds * time.Second)
+	neighbor.client.SetSettings(neighbor.settings)
 	res, err := neighbor.sendRequest(GetBlocks)
 	if err == nil {
 		err = res.GetGob(&blockResponses)
 	}
+	neighbor.settings.SetConnTimeout(commonConnectionTimeoutInSeconds * time.Second)
+	neighbor.client.SetSettings(neighbor.settings)
+	return
+}
+
+func (neighbor *Neighbor) GetLastBlocks(lastBlocksRequest network.LastBlocksRequest) (blockResponses []*network.BlockResponse, err error) {
+	res, err := neighbor.sendRequest(lastBlocksRequest)
+	if err == nil {
+		err = res.GetGob(&blockResponses)
+	}
 	return
 }
 

src/node/protocol/blockchain.go

@@ -5,10 +5,10 @@ import (
 )
 
 type Blockchain interface {
-	Verify(timestamp int64)
+	AddBlock(timestamp int64, transactions []*network.TransactionResponse, registeredAddresses []string)
 	Blocks() []*network.BlockResponse
 	CalculateTotalAmount(currentTimestamp int64, blockchainAddress string) uint64
-	AddBlock(timestamp int64, transactions []*network.TransactionResponse, registeredAddresses []string)
 	Copy() Blockchain
-	IsEmpty() bool
+	LastBlocks(startingBlockHash [32]byte) []*network.BlockResponse
+	Update(timestamp int64)
 }

src/node/protocol/validation/transactions_pool.go

@@ -20,20 +20,18 @@ type TransactionsPool struct {
 	blockchain       protocol.Blockchain
 	registry         protocol.Registry
 	validatorAddress string
-	genesisAmount    uint64
 
 	validationTimer time.Duration
 	watch           clock.Watch
 
 	logger log.Logger
 }
 
-func NewTransactionsPool(blockchain protocol.Blockchain, registry protocol.Registry, validatorAddress string, genesisAmount uint64, validationTimer time.Duration, watch clock.Watch, logger log.Logger) *TransactionsPool {
+func NewTransactionsPool(blockchain protocol.Blockchain, registry protocol.Registry, validatorAddress string, validationTimer time.Duration, watch clock.Watch, logger log.Logger) *TransactionsPool {
 	pool := new(TransactionsPool)
 	pool.blockchain = blockchain
 	pool.registry = registry
 	pool.validatorAddress = validatorAddress
-	pool.genesisAmount = genesisAmount
 	pool.validationTimer = validationTimer
 	pool.watch = watch
 	pool.logger = logger
@@ -61,17 +59,8 @@ func (pool *TransactionsPool) Transactions() []*network.TransactionResponse {
 
 func (pool *TransactionsPool) Validate(timestamp int64) {
 	currentBlockchain := pool.blockchain.Copy()
-	if currentBlockchain.IsEmpty() {
-		genesisTransaction := NewRewardTransaction(pool.validatorAddress, timestamp, pool.genesisAmount)
-		transactions := []*network.TransactionResponse{genesisTransaction}
-		pool.blockchain.AddBlock(timestamp, transactions, nil)
-		pool.logger.Debug("genesis block added")
-		return
-	}
-
 	blocks := currentBlockchain.Blocks()
 	lastBlock := blocks[len(blocks)-1]
-
 	pool.mutex.Lock()
 	defer pool.mutex.Unlock()
 	totalTransactionsValueBySenderAddress := make(map[string]uint64)