improve checkPrerequisites to be able to continue with minimum streams

harmony-one · Aug 24, 2023 · fc07d0c · fc07d0c
1 parent 1ae0bd0
commit fc07d0c
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 2 deletions.
diff --git a/api/service/stagedstreamsync/short_range_helper.go b/api/service/stagedstreamsync/short_range_helper.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/harmony-one/harmony/core/types"
+	"github.com/harmony-one/harmony/internal/utils"
 	syncProto "github.com/harmony-one/harmony/p2p/stream/protocols/sync"
 	sttypes "github.com/harmony-one/harmony/p2p/stream/types"
 	"github.com/pkg/errors"
@@ -132,6 +133,10 @@ func (sh *srHelper) getBlocksByHashes(ctx context.Context, hashes []common.Hash,
 
 func (sh *srHelper) checkPrerequisites() error {
 	if sh.syncProtocol.NumStreams() < sh.config.Concurrency {
+		utils.Logger().Info().
+			Int("available streams", sh.syncProtocol.NumStreams()).
+			Interface("concurrency", sh.config.Concurrency).
+			Msg("not enough streams to do concurrent processes")
 		return ErrNotEnoughStreams
 	}
 	return nil

diff --git a/api/service/stagedstreamsync/stage_epoch.go b/api/service/stagedstreamsync/stage_epoch.go
@@ -92,7 +92,12 @@ func (sr *StageEpoch) doShortRangeSyncForEpochSync(ctx context.Context, s *Stage
 	}
 
 	if err := sh.checkPrerequisites(); err != nil {
-		return 0, errors.Wrap(err, "prerequisite")
+		// if error is ErrNotEnoughStreams but still some streams available,
+		// it can continue syncing, otherwise return error
+		// here we are not doing concurrent processes, so even 1 stream should be enough
+		if err != ErrNotEnoughStreams || s.state.protocol.NumStreams() == 0 {
+			return 0, errors.Wrap(err, "prerequisite")
+		}
 	}
 	curBN := s.state.bc.CurrentBlock().NumberU64()
 	bns := make([]uint64, 0, BlocksPerRequest)

diff --git a/api/service/stagedstreamsync/stage_short_range.go b/api/service/stagedstreamsync/stage_short_range.go
@@ -97,7 +97,12 @@ func (sr *StageShortRange) doShortRangeSync(ctx context.Context, s *StageState)
 	}
 
 	if err := sh.checkPrerequisites(); err != nil {
-		return 0, errors.Wrap(err, "prerequisite")
+		// if error is ErrNotEnoughStreams but still two streams available,
+		// it can continue syncing, otherwise return error
+		// at least 2 streams are needed to do concurrent processes
+		if err != ErrNotEnoughStreams || s.state.protocol.NumStreams() < 2 {
+			return 0, errors.Wrap(err, "prerequisite")
+		}
 	}
 	curBN := sr.configs.bc.CurrentBlock().NumberU64()
 	blkNums := sh.prepareBlockHashNumbers(curBN)