Merge branch 'main' of https://github.com/ethstorage/es-node into dir…

…ectio

ethstorage/data_file.go

@@ -168,7 +168,7 @@ func (df *DataFile) Read(chunkIdx uint64, len int) ([]byte, error) {
 		return nil, fmt.Errorf("chunk not found")
 	}
 	if len > int(df.chunkSize) {
-		return nil, fmt.Errorf(("read too large"))
+		return nil, fmt.Errorf("read too large")
 	}
 	md := make([]byte, len)
 	n, err := df.file.ReadAt(md, HEADER_SIZE+int64(chunkIdx-df.chunkIdxStart)*int64(df.chunkSize))

ethstorage/p2p/host.go

@@ -7,7 +7,7 @@ import (
 	"sync"
 	"time"
 
-	libp2p "github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p"
 	lconf "github.com/libp2p/go-libp2p/config"
 	"github.com/libp2p/go-libp2p/core/connmgr"
 	"github.com/libp2p/go-libp2p/core/host"

ethstorage/p2p/peer_gater.go

@@ -1,8 +1,8 @@
 package p2p
 
 import (
-	log "github.com/ethereum/go-ethereum/log"
-	peer "github.com/libp2p/go-libp2p/core/peer"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/libp2p/go-libp2p/core/peer"
 )
 
 // ConnectionFactor is the factor by which we multiply the connection score.

ethstorage/p2p/peer_params.go

@@ -86,7 +86,7 @@ var DisabledPeerScoreParams = func(blockTime uint64) pubsub.PeerScoreParams {
 }
 
 // PeerScoreParamsByName is a map of name to function that returns a [pubsub.PeerScoreParams] based on the provided [rollup.Config].
-var PeerScoreParamsByName = map[string](func(blockTime uint64) pubsub.PeerScoreParams){
+var PeerScoreParamsByName = map[string]func(blockTime uint64) pubsub.PeerScoreParams{
 	"light": LightPeerScoreParams,
 	"none":  DisabledPeerScoreParams,
 }

ethstorage/p2p/peer_scorer.go

@@ -6,9 +6,9 @@ import (
 	"strconv"
 	"strings"
 
-	log "github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/log"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
-	peer "github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/peer"
 )
 
 type scorer struct {

ethstorage/p2p/protocol/sync_test.go

@@ -18,7 +18,7 @@ import (
 	"time"
 
 	"github.com/detailyang/go-fallocate"
-	ethereum "github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/ethdb"

ethstorage/p2p/protocol/syncclient.go

@@ -303,6 +303,7 @@ func (s *SyncClient) createTask(sid uint64, lastKvIndex uint64) *task {
 	task := task{
 		Contract:       s.storageManager.ContractAddress(),
 		ShardId:        sid,
+		nextIdx:        0,
 		statelessPeers: make(map[peer.ID]struct{}),
 		peers:          make(map[peer.ID]struct{}),
 	}
@@ -418,6 +419,9 @@ func (s *SyncClient) cleanTasks() {
 			t.SubTasks[i].First = min
 			if t.SubTasks[i].done && !exist {
 				t.SubTasks = append(t.SubTasks[:i], t.SubTasks[i+1:]...)
+				if t.nextIdx > i {
+					t.nextIdx--
+				}
 				i--
 			}
 		}
@@ -627,19 +631,22 @@ func (s *SyncClient) assignBlobRangeTasks() {
 	// Iterate over all the tasks and try to find a pending one
 	for _, t := range s.tasks {
 		maxRange := s.syncerParams.MaxRequestSize / ethstorage.ContractToShardManager[t.Contract].MaxKvSize() * 2
-		for _, stask := range t.SubTasks {
-			st := stask
+		subTaskCount := len(t.SubTasks)
+		for idx := 0; idx < subTaskCount; idx++ {
+			pr := s.getIdlePeerForTask(t)
+			if pr == nil {
+				break
+			}
+			t.nextIdx = t.nextIdx % subTaskCount
+			st := t.SubTasks[t.nextIdx]
+			t.nextIdx++
 			if st.done {
 				continue
 			}
 			// Skip any tasks already running
 			if st.isRunning {
 				continue
 			}
-			pr := s.getIdlePeerForTask(t)
-			if pr == nil {
-				continue
-			}
 
 			last := st.next + maxRange
 			if last > st.Last {

ethstorage/p2p/protocol/task.go

@@ -16,6 +16,7 @@ type task struct {
 	Contract      common.Address // Contract address
 	ShardId       uint64         // ShardId
 	SubTasks      []*subTask
+	nextIdx       int
 	healTask      *healTask
 	SubEmptyTasks []*subEmptyTask
 

ethstorage/p2p/topic_params.go

@@ -86,7 +86,7 @@ var DisabledTopicScoreParams = func(blockTime uint64) pubsub.TopicScoreParams {
 }
 
 // TopicScoreParamsByName is a map of name to [pubsub.TopicScoreParams].
-var TopicScoreParamsByName = map[string](func(blockTime uint64) pubsub.TopicScoreParams){
+var TopicScoreParamsByName = map[string]func(blockTime uint64) pubsub.TopicScoreParams{
 	"light": LightTopicScoreParams,
 	"none":  DisabledTopicScoreParams,
 }

ethstorage/prover/kzg_prover_test.go

@@ -130,7 +130,7 @@ func verifyInclusive(trunkIdx uint64, peInput []byte) error {
 
 	dataHash := common.Hash{}
 	copy(dataHash[:], peInput[:24])
-	index := new(big.Int).SetInt64((int64(trunkIdx)))
+	index := new(big.Int).SetInt64(int64(trunkIdx))
 	decodedData := new(big.Int).SetBytes(peInput[64:96])
 
 	h := crypto.Keccak256Hash([]byte("checkInclusive(bytes32,uint256,uint256,bytes)"))

ethstorage/prover/zk_prover.go

@@ -173,7 +173,7 @@ func readMask(publicFile string) (common.Hash, error) {
 	}
 	defer f.Close()
 	var output []string
-	var decoder *json.Decoder = json.NewDecoder(f)
+	var decoder = json.NewDecoder(f)
 	err = decoder.Decode(&output)
 	if err != nil {
 		return common.Hash{}, err

ethstorage/prover/zk_prover_test.go

@@ -131,7 +131,7 @@ func readXIn(buildDir string) (string, error) {
 	}
 	defer f.Close()
 	var input InputPair
-	var decoder *json.Decoder = json.NewDecoder(f)
+	var decoder = json.NewDecoder(f)
 	err = decoder.Decode(&input)
 	if err != nil {
 		return "", err
@@ -162,7 +162,7 @@ func verifyDecodeSample(proof ZKProof, trunkIdx uint64, encodingKey, mask common
 	defer client.Close()
 
 	encodingKeyBN := new(big.Int).SetBytes(encodingKey[:])
-	indexBN := new(big.Int).SetInt64((int64(trunkIdx)))
+	indexBN := new(big.Int).SetInt64(int64(trunkIdx))
 	maskBN := new(big.Int).SetBytes(mask[:])
 
 	h := crypto.Keccak256Hash([]byte("decodeSample(((uint256,uint256),(uint256[2],uint256[2]),(uint256,uint256)),uint256,uint256,uint256)"))