Sync fixes and append queue updates

cmd/utils/flags.go

@@ -120,7 +120,7 @@ var (
 	DBEngineFlag = &cli.StringFlag{
 		Name:  "db.engine",
 		Usage: "Backing database implementation to use ('leveldb' or 'pebble')",
-		Value: "pebble",
+		Value: "leveldb",
 	}
 	KeyStoreDirFlag = DirectoryFlag{
 		Name:  "keystore",

core/core.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"math/big"
 	"sort"
+	"strings"
 	"time"
 
 	"github.com/dominant-strategies/go-quai/common"
@@ -30,9 +31,9 @@ const (
 	c_appendQueueRetryPeriod            = 1       // Time (in seconds) before retrying to append from AppendQueue
 	c_appendQueueThreshold              = 1000    // Number of blocks to load from the disk to ram on every proc of append queue
 	c_processingCache                   = 10      // Number of block hashes held to prevent multi simultaneous appends on a single block hash
-	c_primeRetryThreshold               = 900     // Number of times a block is retry to be appended before eviction from append queue in Prime
-	c_regionRetryThreshold              = 300     // Number of times a block is retry to be appended before eviction from append queue in Region
-	c_zoneRetryThreshold                = 100     // Number of times a block is retry to be appended before eviction from append queue in Zone
+	c_primeRetryThreshold               = 1800    // Number of times a block is retry to be appended before eviction from append queue in Prime
+	c_regionRetryThreshold              = 1200    // Number of times a block is retry to be appended before eviction from append queue in Region
+	c_zoneRetryThreshold                = 600     // Number of times a block is retry to be appended before eviction from append queue in Zone
 	c_maxFutureBlocks                   = 15      // Number of blocks ahead of the current block to be put in the hashNumberList
 	c_appendQueueRetryPriorityThreshold = 5       // If retry counter for a block is less than this number,  then its put in the special list that is tried first to be appended
 	c_appendQueueRemoveThreshold        = 10      // Number of blocks behind the block should be from the current header to be eligble for removal from the append queue
@@ -130,7 +131,11 @@ func (c *Core) InsertChain(blocks types.Blocks) (int, error) {
 			} else if err.Error() != ErrKnownBlock.Error() {
 				log.Info("Append failed.", "hash", block.Hash(), "err", err)
 			}
-			c.removeFromAppendQueue(block)
+			if err != nil && strings.Contains(err.Error(), "connection refused") {
+				log.Error("Append failed because of conenction refused error")
+			} else {
+				c.removeFromAppendQueue(block)
+			}
 		}
 	}
 	return len(blocks), nil
@@ -145,11 +150,12 @@ func (c *Core) procAppendQueue() {
 	for _, hash := range c.appendQueue.Keys() {
 		if value, exist := c.appendQueue.Peek(hash); exist {
 			hashNumber := types.HashAndNumber{Hash: hash.(common.Hash), Number: value.(blockNumberAndRetryCounter).number}
-			if value.(blockNumberAndRetryCounter).retry < c_appendQueueRetryPriorityThreshold {
-				hashNumberPriorityList = append(hashNumberPriorityList, hashNumber)
-			}
 			if hashNumber.Number < c.CurrentHeader().NumberU64()+c_maxFutureBlocks {
-				hashNumberList = append(hashNumberList, hashNumber)
+				if value.(blockNumberAndRetryCounter).retry < c_appendQueueRetryPriorityThreshold {
+					hashNumberPriorityList = append(hashNumberPriorityList, hashNumber)
+				} else {
+					hashNumberList = append(hashNumberList, hashNumber)
+				}
 			}
 		}
 	}
@@ -354,6 +360,10 @@ func (c *Core) WriteBlock(block *types.Block) {
 		nodeCtx := common.NodeLocation.Context()
 		if order == nodeCtx {
 			c.addToAppendQueue(block)
+			parentHeader := c.GetHeader(block.ParentHash(), block.NumberU64()-1)
+			if parentHeader != nil {
+				c.InsertChain([]*types.Block{block})
+			}
 			// If a dom block comes in and we havent appended it yet
 		} else if order < nodeCtx && c.GetHeaderByHash(block.Hash()) == nil {
 			if c.sl.domClient != nil {

core/slice.go

@@ -52,7 +52,6 @@ type Slice struct {
 	quit chan struct{} // slice quit channel
 
 	domClient  *quaiclient.Client
-	domUrl     string
 	subClients []*quaiclient.Client
 
 	wg                    sync.WaitGroup
@@ -80,7 +79,6 @@ func NewSlice(db ethdb.Database, config *Config, txConfig *TxPoolConfig, txLooku
 		config:         chainConfig,
 		engine:         engine,
 		sliceDb:        db,
-		domUrl:         domClientUrl,
 		quit:           make(chan struct{}),
 		badHashesCache: make(map[common.Hash]bool),
 	}
@@ -1070,7 +1068,9 @@ func (sl *Slice) loadLastState() error {
 	}
 	rawdb.DeletePhCache(sl.sliceDb)
 	sl.bestPhKey = rawdb.ReadBestPhKey(sl.sliceDb)
-	sl.miner.worker.LoadPendingBlockBody()
+	if sl.ProcessingState() {
+		sl.miner.worker.LoadPendingBlockBody()
+	}
 	return nil
 }
 

core/tx_pool.go

@@ -1428,14 +1428,14 @@ func (pool *TxPool) promoteExecutables(accounts []common.InternalAddress) []*typ
 			hash := tx.Hash()
 			pool.all.Remove(hash)
 		}
-		log.Debug("Removed old queued transactions", "count", len(forwards))
+		log.Trace("Removed old queued transactions", "count", len(forwards))
 		// Drop all transactions that are too costly (low balance or out of gas)
 		drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas)
 		for _, tx := range drops {
 			hash := tx.Hash()
 			pool.all.Remove(hash)
 		}
-		log.Debug("Removed unpayable queued transactions", "count", len(drops))
+		log.Trace("Removed unpayable queued transactions", "count", len(drops))
 		queuedNofundsMeter.Mark(int64(len(drops)))
 
 		// Gather all executable transactions and promote them
@@ -1446,7 +1446,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.InternalAddress) []*typ
 				promoted = append(promoted, tx)
 			}
 		}
-		log.Debug("Promoted queued transactions", "count", len(promoted))
+		log.Trace("Promoted queued transactions", "count", len(promoted))
 		queuedGauge.Dec(int64(len(readies)))
 
 		// Drop all transactions over the allowed limit

core/worker.go

@@ -277,18 +277,16 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, db ethdb.Databas
 	phBodyCache, _ := lru.New(pendingBlockBodyLimit)
 	worker.pendingBlockBody = phBodyCache
 
-	if headerchain.ProcessingState() {
-		worker.chainHeadSub = worker.hc.SubscribeChainHeadEvent(worker.chainHeadCh)
-	}
-
 	// Sanitize recommit interval if the user-specified one is too short.
 	recommit := worker.config.Recommit
 	if recommit < minRecommitInterval {
 		log.Warn("Sanitizing miner recommit interval", "provided", recommit, "updated", minRecommitInterval)
 		recommit = minRecommitInterval
 	}
 
-	if processingState {
+	nodeCtx := common.NodeLocation.Context()
+	if headerchain.ProcessingState() && nodeCtx == common.ZONE_CTX {
+		worker.chainHeadSub = worker.hc.SubscribeChainHeadEvent(worker.chainHeadCh)
 		worker.wg.Add(1)
 		go worker.asyncStateLoop()
 	}
@@ -366,7 +364,7 @@ func (w *worker) start() {
 
 // stop sets the running status as 0.
 func (w *worker) stop() {
-	if w.hc.ProcessingState() {
+	if w.hc.ProcessingState() && common.NodeLocation.Context() == common.ZONE_CTX {
 		w.chainHeadSub.Unsubscribe()
 	}
 	atomic.StoreInt32(&w.running, 0)

eth/backend.go

@@ -190,8 +190,8 @@ func New(stack *node.Node, config *ethconfig.Config) (*Quai, error) {
 	}
 
 	// Only index bloom if processing state
-	if eth.core.ProcessingState() {
-		eth.bloomIndexer = core.NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms)	
+	if eth.core.ProcessingState() && nodeCtx == common.ZONE_CTX {
+		eth.bloomIndexer = core.NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms)
 		eth.bloomIndexer.Start(eth.Core().Slice().HeaderChain())
 	}
 
@@ -382,8 +382,10 @@ func (s *Quai) Protocols() []p2p.Protocol {
 func (s *Quai) Start() error {
 	eth.StartENRUpdater(s.core, s.p2pServer.LocalNode())
 
-	// Start the bloom bits servicing goroutines
-	s.startBloomHandlers(params.BloomBitsBlocks)
+	if s.core.ProcessingState() && common.NodeLocation.Context() == common.ZONE_CTX {
+		// Start the bloom bits servicing goroutines
+		s.startBloomHandlers(params.BloomBitsBlocks)
+	}
 
 	// Figure out a max peers count based on the server limits
 	maxPeers := s.p2pServer.MaxPeers
@@ -399,9 +401,11 @@ func (s *Quai) Stop() error {
 	s.ethDialCandidates.Close()
 	s.handler.Stop()
 
-	// Then stop everything else.
-	s.bloomIndexer.Close()
-	close(s.closeBloomHandler)
+	if s.core.ProcessingState() && common.NodeLocation.Context() == common.ZONE_CTX {
+		// Then stop everything else.
+		s.bloomIndexer.Close()
+		close(s.closeBloomHandler)
+	}
 	s.core.Stop()
 	s.engine.Close()
 	rawdb.PopUncleanShutdownMarker(s.chainDb)

eth/downloader/downloader.go

@@ -1128,8 +1128,6 @@ func (d *Downloader) processFullSyncContent(peerHeight uint64) error {
 			if err := d.importBlockResults(results); err != nil {
 				return err
 			}
-			d.headNumber = results[len(results)-1].Header.NumberU64()
-			d.headEntropy = d.core.TotalLogS(results[len(results)-1].Header)
 			// If all the blocks are fetched, we exit the sync process
 			if d.headNumber == peerHeight {
 				return errNoFetchesPending
@@ -1152,7 +1150,7 @@ func (d *Downloader) importBlockResults(results []*fetchResult) error {
 	}
 	// Retrieve the a batch of results to import
 	first, last := results[0].Header, results[len(results)-1].Header
-	log.Debug("Inserting downloaded chain", "items", len(results),
+	log.Info("Inserting downloaded chain", "items", len(results),
 		"firstnum", first.Number(), "firsthash", first.Hash(),
 		"lastnum", last.Number(), "lasthash", last.Hash(),
 	)
@@ -1162,6 +1160,8 @@ func (d *Downloader) importBlockResults(results []*fetchResult) error {
 		if d.core.IsBlockHashABadHash(block.Hash()) {
 			return errBadBlockFound
 		}
+		d.headNumber = block.NumberU64()
+		d.headEntropy = d.core.TotalLogS(block.Header())
 		d.core.WriteBlock(block)
 	}
 	return nil

eth/downloader/queue.go

@@ -653,11 +653,7 @@ func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh
 		headers[i], headers[j] = headers[j], headers[i]
 	}
 
-	if len(headers) == 0 && accepted {
-		return 0, nil
-	}
-
-	if accepted {
+	if len(headers) > 0 && accepted {
 		if headers[len(headers)-1].Number().Uint64() != request.From {
 			logger.Info("First header broke chain ordering", "number", headers[0].Number(), "hash", headers[0].Hash(), "expected", request.From)
 			accepted = false
@@ -667,19 +663,19 @@ func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh
 				accepted = false
 			}
 		}
-	}
 
-	if accepted {
-		parentHash := headers[0].Hash()
-		for _, header := range headers[1:] {
-			hash := header.Hash()
-			if parentHash != header.ParentHash() {
-				logger.Warn("Header broke chain ancestry", "number", header.Number(), "hash", hash)
-				accepted = false
-				break
+		if accepted {
+			parentHash := headers[0].Hash()
+			for _, header := range headers[1:] {
+				hash := header.Hash()
+				if parentHash != header.ParentHash() {
+					logger.Warn("Header broke chain ancestry", "number", header.Number(), "hash", hash)
+					accepted = false
+					break
+				}
+				// Set-up parent hash for next round
+				parentHash = hash
 			}
-			// Set-up parent hash for next round
-			parentHash = hash
 		}
 	}
 	// If the batch of headers wasn't accepted, mark as unavailable
@@ -697,7 +693,9 @@ func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh
 		return 0, errors.New("delivery not accepted")
 	}
 
-	copy(q.headerResults[targetTo-q.headerOffset:], headers)
+	if len(headers) > 0 {
+		copy(q.headerResults[targetTo-q.headerOffset:], headers)
+	}
 
 	// Clean up a successful fetch and try to deliver any sub-results
 	delete(q.headerTaskPool, request.From+1)

eth/handler.go

@@ -532,6 +532,12 @@ func (h *handler) missingPendingEtxsLoop() {
 				log.Trace("Fetching the missing pending etxs from", "peer", peer.ID(), "hash", hashAndLocation.Hash)
 				peer.RequestOnePendingEtxs(hashAndLocation.Hash)
 			}
+			if len(peersRunningSlice) == 0 {
+				for _, peer := range h.selectSomePeers() {
+					log.Trace("Fetching the missing pending etxs from", "peer", peer.ID(), "hash", hashAndLocation.Hash)
+					peer.RequestOnePendingEtxs(hashAndLocation.Hash)
+				}
+			}
 		case <-h.missingPendingEtxsSub.Err():
 			return
 		}

eth/sync.go

@@ -30,8 +30,8 @@ import (
 )
 
 const (
-	forceSyncCycle      = 10 * time.Second // Time interval to force syncs, even if few peers are available
-	defaultMinSyncPeers = 1                // Amount of peers desired to start syncing
+	forceSyncCycle      = 60 * time.Second // Time interval to force syncs, even if few peers are available
+	defaultMinSyncPeers = 3                // Amount of peers desired to start syncing
 
 	// This is the target size for the packs of transactions sent by txsyncLoop64.
 	// A pack can get larger than this if a single transactions exceeds this size.
@@ -274,24 +274,25 @@ func (cs *chainSyncer) modeAndLocalHead() (downloader.SyncMode, *big.Int) {
 
 // startSync launches doSync in a new goroutine.
 func (cs *chainSyncer) startSync(op *chainSyncOp) {
-	cs.doneCh = make(chan error, 1)
+	cs.doneCh = make(chan error, 10)
 	go func() { cs.doneCh <- cs.handler.doSync(op) }()
 }
 
 // doSync synchronizes the local blockchain with a remote peer.
 func (h *handler) doSync(op *chainSyncOp) error {
 	// Run the sync cycle, and disable fast sync if we're past the pivot block
 	err := h.downloader.Synchronise(op.peer.ID(), op.head, op.entropy, op.mode)
+	log.Info("Downloader exited", "err", err)
 	if err != nil {
 		return err
 	}
 	// If we've successfully finished a sync cycle and passed any required checkpoint,
 	// enable accepting transactions from the network.
 	head := h.core.CurrentBlock()
-    if head == nil {
-        log.Warn("doSync: head is nil", "hash", h.core.CurrentHeader().Hash(), "number", h.core.CurrentHeader().NumberArray())
-        return nil
-    }
+	if head == nil {
+		log.Warn("doSync: head is nil", "hash", h.core.CurrentHeader().Hash(), "number", h.core.CurrentHeader().NumberArray())
+		return nil
+	}
 	if head.NumberU64() > 0 {
 		// We've completed a sync cycle, notify all peers of new state. This path is
 		// essential in star-topology networks where a gateway node needs to notify

network.env.dist

@@ -126,4 +126,4 @@ ENABLE_PPROF=false
 # Output format variables
 SHOW_COLORS=true
 RUN_BLAKE3=false
-DB_ENGINE=pebble
+DB_ENGINE=leveldb