Added reversed for proof-of-stake syncing

cmd/sentry/download/downloader.go

@@ -353,6 +353,7 @@ func NewControlServer(db kv.RwDB, nodeName string, chainConfig *params.ChainConf
 		1024*1024, /* linkLimit */
 		engine,
 	)
+
 	if err := hd.RecoverFromDb(db); err != nil {
 		return nil, fmt.Errorf("recovery from DB failed: %w", err)
 	}
@@ -470,30 +471,42 @@ func (cs *ControlServerImpl) blockHeaders(ctx context.Context, pkt eth.BlockHead
 			Number:    number,
 		})
 	}
-
 	if segments, penaltyKind, err := cs.Hd.SplitIntoSegments(csHeaders); err == nil {
 		if penaltyKind == headerdownload.NoPenalty {
-			var canRequestMore bool
-			for _, segment := range segments {
-				requestMore, penalties := cs.Hd.ProcessSegment(segment, false /* newBlock */, ConvertH256ToPeerID(peerID))
-				canRequestMore = canRequestMore || requestMore
-				if len(penalties) > 0 {
-					cs.Penalize(ctx, penalties)
+			if cs.Hd.GetBackwards() {
+				tx, err := cs.db.BeginRo(ctx)
+				defer tx.Rollback()
+				if err != nil {
+					return err
 				}
-			}
-
-			if canRequestMore {
-				currentTime := uint64(time.Now().Unix())
-				req, penalties := cs.Hd.RequestMoreHeaders(currentTime)
-				if req != nil {
-					if _, sentToPeer := cs.SendHeaderRequest(ctx, req); sentToPeer {
-						// If request was actually sent to a peer, we update retry time to be 5 seconds in the future
-						cs.Hd.UpdateRetryTime(req, currentTime, 5 /* timeout */)
-						log.Trace("Sent request", "height", req.Number)
+				for _, segment := range segments {
+					if err := cs.Hd.ProcessSegmentPOS(segment, tx); err != nil {
+						return err
+					}
+				}
+			} else {
+				var canRequestMore bool
+				for _, segment := range segments {
+					requestMore, penalties := cs.Hd.ProcessSegment(segment, false /* newBlock */, ConvertH256ToPeerID(peerID))
+					canRequestMore = canRequestMore || requestMore
+					if len(penalties) > 0 {
+						cs.Penalize(ctx, penalties)
 					}
 				}
-				if len(penalties) > 0 {
-					cs.Penalize(ctx, penalties)
+
+				if canRequestMore {
+					currentTime := uint64(time.Now().Unix())
+					req, penalties := cs.Hd.RequestMoreHeaders(currentTime)
+					if req != nil {
+						if _, sentToPeer := cs.SendHeaderRequest(ctx, req); sentToPeer {
+							// If request was actually sent to a peer, we update retry time to be 5 seconds in the future
+							cs.Hd.UpdateRetryTime(req, currentTime, 5 /* timeout */)
+							log.Trace("Sent request", "height", req.Number)
+						}
+					}
+					if len(penalties) > 0 {
+						cs.Penalize(ctx, penalties)
+					}
 				}
 			}
 		} else {

eth/stagedsync/stage_headers.go

@@ -11,6 +11,7 @@ import (
 	"github.com/c2h5oh/datasize"
 	"github.com/holiman/uint256"
 	libcommon "github.com/ledgerwatch/erigon-lib/common"
+	"github.com/ledgerwatch/erigon-lib/etl"
 	"github.com/ledgerwatch/erigon-lib/kv"
 	"github.com/ledgerwatch/erigon/cmd/rpcdaemon/interfaces"
 	"github.com/ledgerwatch/erigon/common"
@@ -38,6 +39,7 @@ type HeadersCfg struct {
 	penalize          func(context.Context, []headerdownload.PenaltyItem)
 	batchSize         datasize.ByteSize
 	noP2PDiscovery    bool
+	tmpdir            string
 	reverseDownloadCh chan types.Header
 	waitingPosHeaders *bool
 	snapshots         *snapshotsync.AllSnapshots
@@ -58,6 +60,7 @@ func StageHeadersCfg(
 	waitingPosHeaders *bool,
 	snapshots *snapshotsync.AllSnapshots,
 	blockReader interfaces.FullBlockReader,
+	tmpdir string,
 ) HeadersCfg {
 	return HeadersCfg{
 		db:                db,
@@ -110,7 +113,7 @@ func SpawnStageHeaders(
 	}
 
 	if isTrans {
-		return HeadersDownward(s, u, ctx, tx, cfg, initialCycle, test)
+		return HeadersDownward(s, u, ctx, tx, cfg, initialCycle, test, useExternalTx)
 	} else {
 		return HeadersForward(s, u, ctx, tx, cfg, initialCycle, test, useExternalTx)
 	}
@@ -125,15 +128,14 @@ func HeadersDownward(
 	cfg HeadersCfg,
 	initialCycle bool,
 	test bool, // Set to true in tests, allows the stage to fail rather than wait indefinitely
+	useExternalTx bool,
 ) error {
 	*cfg.waitingPosHeaders = true
 	// Waiting for the beacon chain
 	log.Info("Waiting for payloads...")
 	header := <-cfg.reverseDownloadCh
 	*cfg.waitingPosHeaders = false
 
-	defer tx.Commit()
-
 	headerNumber := header.Number.Uint64()
 
 	blockHash, err := rawdb.ReadCanonicalHash(tx, headerNumber)
@@ -162,10 +164,97 @@ func HeadersDownward(
 		return err
 	}
 	if parent != nil && parent.Hash() == header.ParentHash {
-		return s.Update(tx, header.Number.Uint64())
+		if err := s.Update(tx, header.Number.Uint64()); err != nil {
+			return err
+		}
+		// For the sake of simplicity we can just assume it will be valid for now. (TODO: move to execution stage)
+		cfg.statusCh <- privateapi.ExecutionStatus{
+			Status:   privateapi.Valid,
+			HeadHash: header.Hash(),
+		}
+		return tx.Commit()
+	}
+	cfg.hd.SetBackwards(true)
+	if err = cfg.hd.ReadProgressFromDb(tx); err != nil {
+		return err
+	}
+	cfg.hd.SetProcessed(header.Number.Uint64())
+	cfg.hd.SetExpectedHash(header.ParentHash)
+	cfg.hd.SetFetching(true)
+	defer cfg.hd.SetFetching(false)
+	logPrefix := s.LogPrefix()
+
+	logEvery := time.NewTicker(logInterval)
+	defer logEvery.Stop()
+
+	// Allow other stages to run 1 cycle if no network available
+	if initialCycle && cfg.noP2PDiscovery {
+		return nil
+	}
+
+	cfg.statusCh <- privateapi.ExecutionStatus{
+		Status:   privateapi.Syncing,
+		HeadHash: rawdb.ReadHeadBlockHash(tx),
 	}
-	// Downward sync if we need to process more (TODO)
-	return s.Update(tx, header.Number.Uint64())
+	log.Info(fmt.Sprintf("[%s] Waiting for headers...", logPrefix), "from", header.Number.Uint64())
+
+	cfg.hd.SetHeaderReader(&chainReader{config: &cfg.chainConfig, tx: tx, blockReader: cfg.blockReader})
+
+	stopped := false
+	prevProgress := header.Number.Uint64()
+	go func() {
+		for {
+			timer := time.NewTimer(1 * time.Second)
+			select {
+			case <-ctx.Done():
+				stopped = true
+			case <-logEvery.C:
+				diff := prevProgress - cfg.hd.Progress()
+				if cfg.hd.Progress() <= prevProgress {
+					log.Info("Wrote Block Headers backwards", "from", header.Number.Uint64(),
+						"now", cfg.hd.Progress(), "blk/sec", float64(diff)/float64(logInterval/time.Second))
+					prevProgress = cfg.hd.Progress()
+				}
+			}
+			timer.Stop()
+		}
+	}()
+	headerCollector := etl.NewCollector(logPrefix, cfg.tmpdir, etl.NewSortableBuffer(etl.BufferOptimalSize))
+	canonicalHeadersCollector := etl.NewCollector(logPrefix, cfg.tmpdir, etl.NewSortableBuffer(etl.BufferOptimalSize))
+	cfg.hd.SetHeadersCollector(headerCollector)
+	cfg.hd.SetCanonicalHashesCollector(canonicalHeadersCollector)
+	defer headerCollector.Close()
+	defer canonicalHeadersCollector.Close()
+	var req headerdownload.HeaderRequest
+	for !stopped {
+		sentToPeer := false
+		for !sentToPeer {
+			req = cfg.hd.RequestMoreHeadersForPOS()
+			_, sentToPeer = cfg.headerReqSend(ctx, &req)
+		}
+		// Load headers into the database
+		announces := cfg.hd.GrabAnnounces()
+		if len(announces) > 0 {
+			cfg.announceNewHashes(ctx, announces)
+		}
+		if cfg.hd.Synced() { // We do not break unless there best header changed
+			break
+		}
+	}
+	if err := headerCollector.Load(tx, kv.Headers, etl.IdentityLoadFunc, etl.TransformArgs{}); err != nil {
+		return err
+	}
+	if err := canonicalHeadersCollector.Load(tx, kv.HeaderCanonical, etl.IdentityLoadFunc, etl.TransformArgs{}); err != nil {
+		return err
+	}
+	if s.BlockNumber >= cfg.hd.Progress() {
+		u.UnwindTo(cfg.hd.Progress(), common.Hash{})
+	} else {
+		if err := s.Update(tx, header.Number.Uint64()); err != nil {
+			return err
+		}
+	}
+	return tx.Commit()
 }
 
 // HeadersForward progresses Headers stage in the forward direction
@@ -261,7 +350,6 @@ func HeadersForward(
 					if header.Eip3675 {
 						return nil
 					}
-
 					if td.Cmp(cfg.terminalTotalDifficulty) > 0 {
 						return rawdb.MarkTransition(tx, blockNum)
 					}
@@ -305,6 +393,7 @@ func HeadersForward(
 	if err = cfg.hd.ReadProgressFromDb(tx); err != nil {
 		return err
 	}
+	cfg.hd.SetBackwards(false)
 	cfg.hd.SetFetching(true)
 	defer cfg.hd.SetFetching(false)
 	headerProgress = cfg.hd.Progress()

ethdb/privateapi/ethbackend.go

@@ -239,7 +239,6 @@ func (s *EthBackendServer) EngineExecutePayloadV1(ctx context.Context, req *type
 	if header.Hash() != blockHash {
 		return nil, fmt.Errorf("invalid hash for payload. got: %s, wanted: %s", common.Bytes2Hex(blockHash[:]), common.Bytes2Hex(header.Hash().Bytes()))
 	}
-	log.Info("Received Payload from beacon-chain", "hash", blockHash)
 	// Send the block over
 	s.numberSent = req.BlockNumber
 	s.reverseDownloadCh <- header

turbo/stages/headerdownload/header_algos.go

@@ -15,6 +15,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/ledgerwatch/erigon-lib/etl"
 	"github.com/ledgerwatch/erigon-lib/kv"
 	"github.com/ledgerwatch/erigon/cmd/rpcdaemon/interfaces"
 	"github.com/ledgerwatch/erigon/common"
@@ -571,6 +572,19 @@ func (hd *HeaderDownload) RequestMoreHeaders(currentTime uint64) (*HeaderRequest
 	return nil, penalties
 }
 
+func (hd *HeaderDownload) RequestMoreHeadersForPOS() HeaderRequest {
+	hd.lock.RLock()
+	defer hd.lock.RUnlock()
+	// Assemble the request
+	return HeaderRequest{
+		Hash:    common.Hash{},
+		Number:  hd.lastProcessedPayload - 1,
+		Length:  192,
+		Skip:    0,
+		Reverse: true,
+	}
+}
+
 func (hd *HeaderDownload) UpdateRetryTime(req *HeaderRequest, currentTime, timeout uint64) {
 	hd.lock.Lock()
 	defer hd.lock.Unlock()
@@ -682,6 +696,48 @@ func (hd *HeaderDownload) InsertHeaders(hf func(header *types.Header, headerRaw
 	return hd.highestInDb >= hd.preverifiedHeight && hd.topSeenHeight > 0 && hd.highestInDb >= hd.topSeenHeight, nil
 }
 
+func (hd *HeaderDownload) SetExpectedHash(hash common.Hash) {
+	hd.lock.Lock()
+	defer hd.lock.Unlock()
+	hd.expectedHash = hash
+}
+
+func (hd *HeaderDownload) ProcessSegmentPOS(segment ChainSegment, tx kv.Getter) error {
+	if len(segment) == 0 {
+		return nil
+	}
+	hd.lock.Lock()
+	defer hd.lock.Unlock()
+	log.Trace("Collecting...", "from", segment[0].Number, "to", segment[len(segment)-1].Number, "len", len(segment))
+	for _, segmentFragment := range segment {
+		header := segmentFragment.Header
+		// If we found the block number we were missing, we can just dismiss it
+		if header.Hash() != hd.expectedHash {
+			return nil
+		}
+		currentCanonical, err := rawdb.ReadCanonicalHash(tx, header.Number.Uint64())
+		if err != nil {
+			return err
+		}
+		if currentCanonical == hd.expectedHash || hd.lastProcessedPayload == 1 {
+			hd.synced = true
+		}
+		data, err := rlp.EncodeToBytes(header)
+		if err != nil {
+			log.Crit("Failed to RLP encode header", "err", err)
+		}
+		if err := hd.headersCollector.Collect(dbutils.HeaderKey(header.Number.Uint64(), header.Hash()), data); err != nil {
+			return err
+		}
+		if err := hd.canonicalHashesCollector.Collect(dbutils.EncodeBlockNumber(header.Number.Uint64()), header.Hash().Bytes()); err != nil {
+			return err
+		}
+		hd.expectedHash = segmentFragment.Header.ParentHash
+		hd.lastProcessedPayload = header.Number.Uint64()
+	}
+	return nil
+}
+
 // GrabAnnounces - returns all available announces and forget them
 func (hd *HeaderDownload) GrabAnnounces() []Announce {
 	hd.lock.Lock()
@@ -694,7 +750,11 @@ func (hd *HeaderDownload) GrabAnnounces() []Announce {
 func (hd *HeaderDownload) Progress() uint64 {
 	hd.lock.RLock()
 	defer hd.lock.RUnlock()
-	return hd.highestInDb
+	if hd.backwards {
+		return hd.lastProcessedPayload
+	} else {
+		return hd.highestInDb
+	}
 }
 
 func (hd *HeaderDownload) HasLink(linkHash common.Hash) bool {
@@ -965,6 +1025,42 @@ func (hd *HeaderDownload) SetFetching(fetching bool) {
 	hd.fetching = fetching
 }
 
+func (hd *HeaderDownload) SetProcessed(lastProcessed uint64) {
+	hd.lock.Lock()
+	defer hd.lock.Unlock()
+	hd.lastProcessedPayload = lastProcessed
+}
+
+func (hd *HeaderDownload) SetHeadersCollector(collector *etl.Collector) {
+	hd.lock.Lock()
+	defer hd.lock.Unlock()
+	hd.headersCollector = collector
+}
+
+func (hd *HeaderDownload) SetCanonicalHashesCollector(collector *etl.Collector) {
+	hd.lock.Lock()
+	defer hd.lock.Unlock()
+	hd.canonicalHashesCollector = collector
+}
+
+func (hd *HeaderDownload) SetBackwards(backwards bool) {
+	hd.lock.Lock()
+	defer hd.lock.Unlock()
+	hd.backwards = backwards
+}
+
+func (hd *HeaderDownload) GetBackwards() bool {
+	hd.lock.RLock()
+	defer hd.lock.RUnlock()
+	return hd.backwards
+}
+
+func (hd *HeaderDownload) Synced() bool {
+	hd.lock.RLock()
+	defer hd.lock.RUnlock()
+	return hd.synced
+}
+
 func (hd *HeaderDownload) RequestChaining() bool {
 	hd.lock.RLock()
 	defer hd.lock.RUnlock()

turbo/stages/headerdownload/header_data_struct.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 
 	lru "github.com/hashicorp/golang-lru"
+	"github.com/ledgerwatch/erigon-lib/etl"
 	"github.com/ledgerwatch/erigon/common"
 	"github.com/ledgerwatch/erigon/consensus"
 	"github.com/ledgerwatch/erigon/core/types"
@@ -219,6 +220,13 @@ type HeaderDownload struct {
 	topSeenHeight      uint64
 	requestChaining    bool // Whether the downloader is allowed to issue more requests when previous responses created or moved an anchor
 	fetching           bool // Set when the stage that is actively fetching the headers is in progress
+	// proof-of-stake
+	lastProcessedPayload     uint64         // The last header number inserted when processing the chain backwards
+	expectedHash             common.Hash    // Parenthash of the last header inserted, we keep it so that we do not read it from database over and over
+	synced                   bool           // if we found a canonical hash during backward sync, in this case our sync process is done
+	backwards                bool           // True if the chain is syncing backwards or not
+	headersCollector         *etl.Collector // ETL collector for headers
+	canonicalHashesCollector *etl.Collector // ETL collector for canonical hashes
 }
 
 // HeaderRecord encapsulates two forms of the same header - raw RLP encoding (to avoid duplicated decodings and encodings), and parsed value types.Header