core: unembedded functions, improved and readable block queue logic

core/index_server.go

@@ -18,6 +18,7 @@
 package core
 
 import (
+	"math"
 	"sync"
 	"time"
 
@@ -132,8 +133,8 @@ func (f *indexServers) register(indexer Indexer, name string) {
 	indexer.SetHistoryCutoff(f.historyCutoff)
 	indexer.SetFinalized(f.finalBlock)
 	if f.lastHead != nil {
-		server.status.ready, server.status.needBlocks = indexer.AddBlockData(f.lastHead, f.lastHeadReceipts)
+		server.sendStatus.ready, server.sendStatus.needBlocks = indexer.AddBlockData(f.lastHead, f.lastHeadReceipts)
-		server.updateStatus()
+		server.updateSendStatus()
 	}
 	go server.historicReadLoop()
 	go server.historicSendLoop()
@@ -227,15 +228,17 @@ type indexServer struct {
 	stopped bool
 
 	lastHead                          *types.Header
-	status                            indexerStatus
+	sendStatus                        indexerStatus
 	statusCh                          chan indexerStatus
 	blockDataCh                       chan blockData
 	suspendCh                         chan bool
 	testSuspendHookCh                 chan struct{} // initialized by test, capacity = 1
-	lastRevertBlock                   uint64
 	sendTimer                         *time.Timer
 	historyCutoff, missingBlockCutoff uint64
 
+	readStatus  indexerStatus
+	readPointer uint64 // next block to be queued
+
 	name                    string
 	processed               uint64
 	logged                  bool
@@ -243,12 +246,35 @@ type indexServer struct {
 	lastHistoryErrorLog     time.Time
 }
 
-// indexerStatus represents the state of the indexer and also has fields that
+// indexerStatus is maintained by the historicSendLoop goroutine and all changes
-// serve the coordination between historic reader and sender goroutines.
+// are sent to the historicReadLoop goroutine through statusCh.
 type indexerStatus struct {
-	ready, suspended, resetQueue bool
+	ready                        bool                 // last feedback received from the indexer
-	revertCount                  uint64
+	needBlocks                   common.Range[uint64] // last feedback received from the indexer
-	needBlocks                   common.Range[uint64]
+	suspended                    bool                 // suspend historic block delivery during block processing
+	resetQueueCount              uint64               // total number of queue resets
+	revertCount, lastRevertBlock uint64               // detect entries potentially expired by a revert/reorg
+}
+
+// isNextExpected returns true if the received blockData (potentially based on a
+// previously sent indexerStatus) is still guaranteed to be valid and the one
+// expected by the indexer according to the latest indexerStatus.
+func (s *indexerStatus) isNextExpected(b blockData) bool {
+	if s.needBlocks.IsEmpty() || s.needBlocks.First() != b.blockNumber {
+		return false // not the expected next block number or no historical blocks expected at all
+	}
+	// block number is the expected one; check if a reorg might have invalidated it
+	switch s.revertCount {
+	case b.revertCount:
+		return true // no reorgs happened since the collection of block data
+	case b.revertCount + 1:
+		// one reorg happened to s.lastRevertBlock; b is valid if not newer than this
+		return b.blockNumber <= s.lastRevertBlock
+	default:
+		// multiple reorgs happened; previous revert blocks are not remembered
+		// so we don't know if b is still valid and therefore we have to discard it.
+		return false
+	}
 }
 
 // blockData represents the indexable data of a single block being sent from the
@@ -277,8 +303,22 @@ func (s *indexServer) sendHeadBlockData(header *types.Header, receipts types.Rec
 		return
 	}
 	s.lastHead = header
-	s.status.ready, s.status.needBlocks = s.indexer.AddBlockData(header, receipts)
+	ready, needBlocks := s.indexer.AddBlockData(header, receipts)
-	s.updateStatus()
+	s.updateIndexerStatus(ready, needBlocks, 0)
+	s.updateSendStatus()
+}
+
+// updateIndexerStatus updates the ready / needBlocks fields in the send loop
+// status. The number of historic blocks added since the last update is
+// specified in addedBlocks. During continuous historical block range delivery
+// the starting point of the new needBlocks range is expected to advance with
+// each new block added. If the new range does not match the expectation then
+// a blockDataCh queue reset is requested.
+func (s *indexServer) updateIndexerStatus(ready bool, needBlocks common.Range[uint64], addedBlocks uint64) {
+	if needBlocks.First() != s.sendStatus.needBlocks.First()+addedBlocks {
+		s.sendStatus.resetQueueCount++ // request queue reset
+	}
+	s.sendStatus.ready, s.sendStatus.needBlocks = ready, needBlocks
 }
 
 // revert immediately reverts the indexer to the given block and updates the
@@ -298,12 +338,39 @@ func (s *indexServer) revert(header *types.Header) {
 		panic("invalid indexer revert")
 	}
 	s.lastHead = header
-	s.status.revertCount++
+	s.sendStatus.revertCount++
-	s.lastRevertBlock = blockNumber
+	s.sendStatus.lastRevertBlock = blockNumber
-	s.updateStatus()
+	s.updateSendStatus()
 	s.indexer.Revert(blockNumber)
 }
 
+// suspendOrStop blocks the send loop until it is unsuspended or the parent
+// chain is stopped. It also notifies the indexer by calling Suspend and
+// suspends the read loop through updateStatus.
+func (s *indexServer) suspendOrStop(suspended bool) bool {
+	if !suspended {
+		panic("unexpected 'false' signal on suspendCh")
+	}
+	s.lock.Lock()
+	s.sendStatus.suspended = true
+	s.updateSendStatus()
+	s.indexer.Suspended()
+	s.lock.Unlock()
+	select {
+	case <-s.parent.closeCh:
+		return true
+	case suspended = <-s.suspendCh:
+	}
+	if suspended {
+		panic("unexpected 'true' signal on suspendCh")
+	}
+	s.lock.Lock()
+	s.sendStatus.suspended = false
+	s.updateSendStatus()
+	s.lock.Unlock()
+	return false
+}
+
 // historicSendLoop is the main event loop that interacts with the indexer in
 // case when historical block data is requested. It sends status updates to
 // the reader goroutine through statusCh and feeds the fetched data coming from
@@ -317,40 +384,13 @@ func (s *indexServer) historicSendLoop() {
 		s.parent.closeWg.Done()
 	}()
 
-	// suspendOrStop blocks the send loop until it is unsuspended or the parent
-	// chain is stopped. It also notifies the indexer by calling Suspend and
-	// suspends the read loop through updateStatus.
-	suspendOrStop := func(suspended bool) bool {
-		if !suspended {
-			panic("unexpected 'false' signal on suspendCh")
-		}
-		s.lock.Lock()
-		s.status.suspended = true
-		s.updateStatus()
-		s.indexer.Suspended()
-		s.lock.Unlock()
-		select {
-		case <-s.parent.closeCh:
-			return true
-		case suspended = <-s.suspendCh:
-		}
-		if suspended {
-			panic("unexpected 'true' signal on suspendCh")
-		}
-		s.lock.Lock()
-		s.status.suspended = false
-		s.updateStatus()
-		s.lock.Unlock()
-		return false
-	}
-
 	for {
 		select {
 		// do a separate non-blocking select to ensure that a suspend attempt
 		// during the previous historical AddBlockData will be catched in the
 		// next round.
-		case ch := <-s.suspendCh:
+		case suspend := <-s.suspendCh:
-			if suspendOrStop(ch) {
+			if s.suspendOrStop(suspend) {
 				return
 			}
 		default:
@@ -358,26 +398,24 @@ func (s *indexServer) historicSendLoop() {
 		select {
 		case <-s.parent.closeCh:
 			return
-		case ch := <-s.suspendCh:
+		case suspend := <-s.suspendCh:
-			if suspendOrStop(ch) {
+			if s.suspendOrStop(suspend) {
 				return
 			}
 		case nextBlockData := <-s.blockDataCh:
 			s.lock.Lock()
-			s.status.resetQueue = true
 			// check if received block data is indeed from the next expected
 			// block and is still guaranteed to be canonical; ignore and request
-			// again otherwise.
+			// a queue reset otherwise.
-			if !s.status.needBlocks.IsEmpty() && s.status.needBlocks.First() == nextBlockData.blockNumber &&
+			if s.sendStatus.isNextExpected(nextBlockData) {
-				(nextBlockData.revertCount == s.status.revertCount || (nextBlockData.revertCount+1 == s.status.revertCount && nextBlockData.blockNumber <= s.lastRevertBlock)) {
+				ready, needBlocks := s.indexer.AddBlockData(nextBlockData.header, nextBlockData.receipts)
-				s.status.ready, s.status.needBlocks = s.indexer.AddBlockData(nextBlockData.header, nextBlockData.receipts)
+				s.updateIndexerStatus(ready, needBlocks, 1)
 				// check if the has actually been found in the database
 				if nextBlockData.header != nil && nextBlockData.receipts != nil {
-					s.status.resetQueue = false
+					if s.sendStatus.needBlocks.IsEmpty() {
-					if s.status.needBlocks.IsEmpty() {
 						s.logDelivered(nextBlockData.blockNumber)
 						s.logFinished()
-					} else if s.status.needBlocks.First() == nextBlockData.blockNumber+1 {
+					} else if s.sendStatus.needBlocks.First() == nextBlockData.blockNumber+1 {
 						s.logDelivered(nextBlockData.blockNumber)
 					}
 				} else {
@@ -393,16 +431,21 @@ func (s *indexServer) historicSendLoop() {
 					}
 					s.missingBlockCutoff = max(s.missingBlockCutoff, nextBlockData.blockNumber+1)
 					s.indexer.SetHistoryCutoff(max(s.historyCutoff, s.missingBlockCutoff))
-					s.status.ready, s.status.needBlocks = s.indexer.Status()
+					ready, needBlocks := s.indexer.Status()
+					s.updateIndexerStatus(ready, needBlocks, 0)
 				}
+			} else {
+				// trigger resetting the queue and sending blockData from needBlocks.First()
+				s.sendStatus.resetQueueCount++
 			}
-			s.updateStatus()
+			s.updateSendStatus()
 			s.lock.Unlock()
 		case <-s.sendTimer.C:
 			s.lock.Lock()
-			if !s.status.ready {
+			if !s.sendStatus.ready {
-				s.status.ready, s.status.needBlocks = s.indexer.Status()
+				ready, needBlocks := s.indexer.Status()
-				s.updateStatus()
+				s.updateIndexerStatus(ready, needBlocks, 0)
+				s.updateSendStatus()
 			}
 			s.lock.Unlock()
 		}
@@ -413,8 +456,8 @@ func (s *indexServer) historicSendLoop() {
 // latest indexer status. If necessary then it trims the needBlocks range based
 // on the locally available block range. If there is already an unread status
 // update waiting on statusCh then it is replaced by the new one.
-func (s *indexServer) updateStatus() {
+func (s *indexServer) updateSendStatus() {
-	if s.status.ready || s.status.suspended {
+	if s.sendStatus.ready || s.sendStatus.suspended {
 		s.sendTimer.Stop()
 	} else {
 		s.sendTimer.Reset(busyDelay)
@@ -423,17 +466,17 @@ func (s *indexServer) updateStatus() {
 	if s.lastHead != nil {
 		headNumber = s.lastHead.Number.Uint64()
 	}
-	if headNumber+1 < s.status.needBlocks.AfterLast() {
+	if headNumber+1 < s.sendStatus.needBlocks.AfterLast() {
-		s.status.needBlocks.SetLast(headNumber)
+		s.sendStatus.needBlocks.SetLast(headNumber)
 	}
-	if s.status.needBlocks.IsEmpty() || max(s.historyCutoff, s.missingBlockCutoff) > s.status.needBlocks.First() {
+	if s.sendStatus.needBlocks.IsEmpty() || max(s.historyCutoff, s.missingBlockCutoff) > s.sendStatus.needBlocks.First() {
-		s.status.needBlocks = common.Range[uint64]{}
+		s.sendStatus.needBlocks = common.Range[uint64]{}
 	}
 	select {
 	case <-s.statusCh:
 	default:
 	}
-	s.statusCh <- s.status
+	s.statusCh <- s.sendStatus
 }
 
 // setBlockProcessing suspends serving historical blocks requested by the indexer
@@ -457,6 +500,41 @@ func (s *indexServer) setBlockProcessing(suspended bool) {
 	}
 }
 
+// clearBlockQueue removes all entries from blockDataCh.
+func (s *indexServer) clearBlockQueue() {
+	for {
+		select {
+		case <-s.blockDataCh:
+		default:
+			return
+		}
+	}
+}
+
+// updateReadStatus updates readStatus and checks whether a queue reset has been
+// requested by the send loop. In that case it empties the queue and resets the
+// readPointer to the first block of the needBlocks range.
+// Note that the blocks betweeen needBlocks.First() and readPointer-1 are assumed
+// to already be queued in blockDataCh. If needBlocks.First() does not advance
+// with each delivered block or an expired blockData is received by the send
+// loop then a queue reset is requested.
+func (s *indexServer) updateReadStatus(newStatus indexerStatus) {
+	if newStatus.resetQueueCount != s.readStatus.resetQueueCount {
+		s.clearBlockQueue()
+		s.readPointer = newStatus.needBlocks.First()
+	}
+	s.readStatus = newStatus
+}
+
+// canQueueNextBlock returns true if there are more blocks to read in the
+// needBlocks range and we have not yet reached the capacity of blockDataCh yet.
+// Note that the latter check assumes that blocks between needBlocks.First() and
+// readPointer-1 are queued.
+func (s *indexServer) canQueueNextBlock() bool {
+	return s.readStatus.needBlocks.Includes(s.readPointer) &&
+		s.readPointer <= s.readStatus.needBlocks.First()+maxHistoricPrefetch
+}
+
 // historicReadLoop reads requested historical block data asynchronously.
 // It receives indexer status updates on statusCh and sends block data to
 // blockDataCh. If the latest status indicates that there the server is not
@@ -466,58 +544,24 @@ func (s *indexServer) setBlockProcessing(suspended bool) {
 // Note that blockDataCh can queue multiple block data pre-fetched by
 // historicReadLoop. If the requested range is changed while there is still
 // queued data in the channel that corresponds to the previous requested range
-// then the receiver sends a new status update with the resetQueue flag set to
+// then the receiver sends a new status update with increased resetQueueCount.
-// true. In this case historicReadLoop removes all remaining entries from the
+// In this case historicReadLoop removes all remaining entries from the queue
-// queue and starts sending block data from the beginning of the new range.
+// and starts sending block data from the beginning of the new range.
 func (s *indexServer) historicReadLoop() {
 	defer s.parent.closeWg.Done()
 
-	var (
+	s.readStatus.resetQueueCount = math.MaxUint64
-		status    indexerStatus
-		sendRange common.Range[uint64]
-	)
-
-	statusUpdated := func() {
-		if status.resetQueue {
-			// If the receiver found an item in the queue that is no longer
-			// relevant then we remove all remaining items first.
-		loop:
-			for {
-				select {
-				case <-s.blockDataCh:
-				default:
-					break loop
-				}
-			}
-		}
-		if !status.resetQueue && !sendRange.IsEmpty() && status.needBlocks.Includes(sendRange.First()) {
-			// Here we assume that the block data between needBlocks.First() and
-			// sendRange.First()-1 is already in the queue.
-			r := status.needBlocks
-			r.SetFirst(sendRange.First())
-			sendRange = r
-		} else {
-			sendRange = status.needBlocks
-		}
-		if sendRange.Count() > maxHistoricPrefetch {
-			// Note: in a normal use case where needBlocks.First() is advanced
-			// after reading the previous item from blockDataCh, this check will
-			// prevent reading more data than what fits into the channel capacity.
-			sendRange.SetAfterLast(sendRange.First() + maxHistoricPrefetch)
-		}
-	}
-
 	for {
-		if !sendRange.IsEmpty() && !status.suspended {
+		if !s.readStatus.suspended && s.canQueueNextBlock() {
 			// Send next item to the queue.
-			bd := blockData{blockNumber: sendRange.First(), revertCount: status.revertCount}
+			bd := blockData{blockNumber: s.readPointer, revertCount: s.readStatus.revertCount}
 			if bd.header = s.parent.chain.GetHeaderByNumber(bd.blockNumber); bd.header != nil {
 				blockHash := bd.header.Hash()
 				bd.receipts, _ = s.parent.rawReceiptsCache.Get(blockHash)
 				if bd.receipts == nil {
-					bd.receipts = s.parent.chain.GetRawReceipts(blockHash, bd.blockNumber)
 					// Note: we do not cache historical receipts because typically
 					// each indexer requests them at different times.
+					bd.receipts = s.parent.chain.GetRawReceipts(blockHash, bd.blockNumber)
 				}
 			}
 			// Note that a response with missing block data is still sent in case of
@@ -525,23 +569,26 @@ func (s *indexServer) historicReadLoop() {
 			// This might be either due to a database error or a reorg.
 			select {
 			case s.blockDataCh <- bd:
-				sendRange.SetFirst(bd.blockNumber + 1)
+				s.readPointer++
 			default:
-				// Note: in extreme corner cases where sendRange.Count() check
+				// Note: updateIndexerStatus in the send loop and canQueueNextBlock
-				// does not prevent trying to overfill the channel, we simply
+				// in the read loop should ensure that no send is attempted at
-				// reset sendRange to empty preventing more wasted reads.
+				// blockDataCh when it is filled to full capacity. If it happens
-				// Sending the queued data will generate more status updates
+				// anyway then we print an error and set the suspended flag to
-				// which will reinitialize sendRange once the queue is again
+				// true until the next status update.
-				// below full capacity.
+				if time.Since(s.lastHistoryErrorLog) >= time.Second*10 {
-				sendRange = common.Range[uint64]{}
+					s.lastHistoryErrorLog = time.Now()
+					log.Error("Historical block queue is full")
+				}
+				s.readStatus.suspended = true
 			}
 			// Keep checking status updates without blocking as long as there is
 			// something to do.
 			select {
 			case <-s.parent.closeCh:
 				return
-			case status = <-s.statusCh:
+			case status := <-s.statusCh:
-				statusUpdated()
+				s.updateReadStatus(status)
 			default:
 			}
 		} else {
@@ -549,8 +596,8 @@ func (s *indexServer) historicReadLoop() {
 			select {
 			case <-s.parent.closeCh:
 				return
-			case status = <-s.statusCh:
+			case status := <-s.statusCh:
-				statusUpdated()
+				s.updateReadStatus(status)
 			}
 		}
 	}
@@ -615,6 +662,7 @@ func (s *indexServer) setHistoryCutoff(blockNumber uint64) {
 	}
 	s.historyCutoff = blockNumber
 	s.indexer.SetHistoryCutoff(max(s.historyCutoff, s.missingBlockCutoff))
-	s.status.ready, s.status.needBlocks = s.indexer.Status()
+	ready, needBlocks := s.indexer.Status()
-	s.updateStatus()
+	s.updateIndexerStatus(ready, needBlocks, 0)
+	s.updateSendStatus()
 }