eth/downloader: implement beacon sync

core/rawdb/schema.go

@@ -95,7 +95,7 @@ var (
 	SnapshotAccountPrefix = []byte("a") // SnapshotAccountPrefix + account hash -> account trie value
 	SnapshotStoragePrefix = []byte("o") // SnapshotStoragePrefix + account hash + storage hash -> storage trie value
 	CodePrefix            = []byte("c") // CodePrefix + code hash -> account code
-	skeletonHeaderPrefix  = []byte("S") // skeletonHeaderPrefox + num (uint64 big endian) -> header
+	skeletonHeaderPrefix  = []byte("S") // skeletonHeaderPrefix + num (uint64 big endian) -> header
 
 	PreimagePrefix = []byte("secure-key-")      // PreimagePrefix + hash -> preimage
 	configPrefix   = []byte("ethereum-config-") // config prefix for the db

eth/downloader/downloader.go

@@ -476,12 +476,9 @@ func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td, ttd *
 		if err != nil {
 			return err
 		}
-		// Opposed to legacy mode, in beacon mode we trust the chain we've been
-		// told to sync to, so no need to leave a gap between the pivot and head
-		// to full sync. Still, the downloader's been architected to do a full
-		// block import after the pivot, so make it off by one to avoid having
-		// to special case everything internally.
-		pivot = d.skeleton.Header(latest.Number.Uint64() - 1)
+		if latest.Number.Uint64() > uint64(fsMinFullBlocks) {
+			pivot = d.skeleton.Header(latest.Number.Uint64() - uint64(fsMinFullBlocks))
+		}
 	}
 	// If no pivot block was returned, the head is below the min full block
 	// threshold (i.e. new chain). In that case we won't really snap sync
@@ -1346,18 +1343,31 @@ func (d *Downloader) processHeaders(origin uint64, td, ttd *big.Int, beaconMode
 					// Although the received headers might be all valid, a legacy
 					// PoW/PoA sync must not accept post-merge headers. Make sure
 					// that any transition is rejected at this point.
+					var (
+						rejected []*types.Header
+						td       *big.Int
+					)
 					if !beaconMode && ttd != nil {
-						ptd := d.lightchain.GetTd(chunkHeaders[0].ParentHash, chunkHeaders[0].Number.Uint64()-1)
-						if ptd == nil {
+						td = d.blockchain.GetTd(chunkHeaders[0].ParentHash, chunkHeaders[0].Number.Uint64()-1)
+						if td == nil {
 							// This should never really happen, but handle gracefully for now
 							log.Error("Failed to retrieve parent header TD", "number", chunkHeaders[0].Number.Uint64()-1, "hash", chunkHeaders[0].ParentHash)
 							return fmt.Errorf("%w: parent TD missing", errInvalidChain)
 						}
-						for _, header := range chunkHeaders {
-							ptd = new(big.Int).Add(ptd, header.Difficulty)
-							if ptd.Cmp(ttd) >= 0 {
-								log.Info("Legacy sync reached merge threshold", "number", header.Number, "hash", header.Hash(), "td", ptd, "ttd", ttd)
-								return ErrMergeTransition
+						for i, header := range chunkHeaders {
+							td = new(big.Int).Add(td, header.Difficulty)
+							if td.Cmp(ttd) >= 0 {
+								// Terminal total difficulty reached, allow the last header in
+								if new(big.Int).Sub(td, header.Difficulty).Cmp(ttd) < 0 {
+									chunkHeaders, rejected = chunkHeaders[:i+1], chunkHeaders[i+1:]
+									if len(rejected) > 0 {
+										// Make a nicer user log as to the first TD truly rejected
+										td = new(big.Int).Add(td, rejected[0].Difficulty)
+									}
+								} else {
+									chunkHeaders, rejected = chunkHeaders[:i], chunkHeaders[i:]
+								}
+								break
 							}
 						}
 					}
@@ -1380,6 +1390,13 @@ func (d *Downloader) processHeaders(origin uint64, td, ttd *big.Int, beaconMode
 							rollback = 1
 						}
 					}
+					if len(rejected) != 0 {
+						// Merge threshold reached, stop importing, but don't roll back
+						rollback = 0
+
+						log.Info("Legacy sync reached merge threshold", "number", rejected[0].Number, "hash", rejected[0].Hash(), "td", td, "ttd", ttd)
+						return ErrMergeTransition
+					}
 				}
 				// Unless we're doing light chains, schedule the headers for associated content retrieval
 				if mode == FullSync || mode == SnapSync {

eth/downloader/skeleton.go

@@ -177,7 +177,7 @@ type backfiller interface {
 // concurrently with the sync cycle, since extensions arrive from an API surface,
 // not from within (vs. legacy Ethereum sync).
 //
-// Since the skeleton tracks the entire header chain until it is cosumed by the
+// Since the skeleton tracks the entire header chain until it is consumed by the
 // forward block filling, it needs 0.5KB/block storage. At current mainnet sizes
 // this is only possible with a disk backend. Since the skeleton is separate from
 // the node's header chain, storing the headers ephemerally until sync finishes
@@ -748,13 +748,13 @@ func (s *skeleton) executeTask(peer *peerConnection, req *headerRequest) {
 			res.Done <- errors.New("invalid header batch anchor")
 			s.scheduleRevertRequest(req)
 
-		case headers[0].Number.Uint64() >= requestHeaders && len(headers) != requestHeaders:
+		case req.head >= requestHeaders && len(headers) != requestHeaders:
 			// Invalid number of non-genesis headers delivered, reject the response and reschedule
 			peer.log.Debug("Invalid non-genesis header count", "have", len(headers), "want", requestHeaders)
 			res.Done <- errors.New("not enough non-genesis headers delivered")
 			s.scheduleRevertRequest(req)
 
-		case headers[0].Number.Uint64() < requestHeaders && uint64(len(headers)) != headers[0].Number.Uint64():
+		case req.head < requestHeaders && uint64(len(headers)) != req.head:
 			// Invalid number of genesis headers delivered, reject the response and reschedule
 			peer.log.Debug("Invalid genesis header count", "have", len(headers), "want", headers[0].Number.Uint64())
 			res.Done <- errors.New("not enough genesis headers delivered")
@@ -953,6 +953,12 @@ func (s *skeleton) processResponse(res *headerResponse) bool {
 				merged = true
 			}
 		}
+		// If subchains were merged, all further available headers in the scratch
+		// space are invalid since we skipped ahead. Stop processing the scratch
+		// space to avoid dropping peers thinking they delivered invalid data.
+		if merged {
+			break
+		}
 	}
 	s.saveSyncStatus(batch)
 	if err := batch.Write(); err != nil {

eth/downloader/skeleton_test.go

@@ -79,20 +79,33 @@ func (hf *hookedBackfiller) resume() {
 type skeletonTestPeer struct {
 	id      string          // Unique identifier of the mock peer
 	headers []*types.Header // Headers to serve when requested
-	served  uint64          // Number of headers served by this peer
-	dropped uint64          // Flag whether the peer was dropped (stop responding)
+
+	serve func(origin uint64) []*types.Header // Hook to allow custom responses
+
+	served  uint64 // Number of headers served by this peer
+	dropped uint64 // Flag whether the peer was dropped (stop responding)
 }
 
 // newSkeletonTestPeer creates a new mock peer to test the skeleton sync with.
-// The only purpose of the constructor is to ensure we don't forget to set some
-// mandatory field vs a struct literal initialization.
 func newSkeletonTestPeer(id string, headers []*types.Header) *skeletonTestPeer {
 	return &skeletonTestPeer{
 		id:      id,
 		headers: headers,
 	}
 }
 
+// newSkeletonTestPeer creates a new mock peer to test the skeleton sync with,
+// and sets an optional serve hook that can return headers for delivery instead
+// of the predefined chain. Useful for emulating malicious behavior that would
+// otherwise require dedicated peer types.
+func newSkeletonTestPeerWithHook(id string, headers []*types.Header, serve func(origin uint64) []*types.Header) *skeletonTestPeer {
+	return &skeletonTestPeer{
+		id:      id,
+		headers: headers,
+		serve:   serve,
+	}
+}
+
 // RequestHeadersByNumber constructs a GetBlockHeaders function based on a numbered
 // origin; associated with a particular peer in the download tester. The returned
 // function can be used to retrieve batches of headers from the particular peer.
@@ -126,18 +139,26 @@ func (p *skeletonTestPeer) RequestHeadersByNumber(origin uint64, amount int, ski
 	if amount > requestHeaders || (amount < requestHeaders && origin > uint64(amount)) {
 		panic(fmt.Sprintf("non-chunk size header batch requested: requested %d, want %d, origin %d", amount, requestHeaders, origin))
 	}
-	// Simple reverse header retrieval. Fill from the peer's chain and return
-	headers := make([]*types.Header, 0, amount)
-	if len(p.headers) > int(origin) { // Don't serve headers if we're missing the origin
-		for i := 0; i < amount; i++ {
-			// Consider nil headers as a form of attack and withhold them. Nil
-			// cannot be decoded from RLP, so it's not possible to produce an
-			// attack by sending/receiving those over eth.
-			header := p.headers[int(origin)-i]
-			if header == nil {
-				continue
+	// Simple reverse header retrieval. Fill from the peer's chain and return.
+	// If the tester has a serve hook set, try to use that before falling back
+	// to the default behavior.
+	var headers []*types.Header
+	if p.serve != nil {
+		headers = p.serve(origin)
+	}
+	if headers == nil {
+		headers = make([]*types.Header, 0, amount)
+		if len(p.headers) > int(origin) { // Don't serve headers if we're missing the origin
+			for i := 0; i < amount; i++ {
+				// Consider nil headers as a form of attack and withhold them. Nil
+				// cannot be decoded from RLP, so it's not possible to produce an
+				// attack by sending/receiving those over eth.
+				header := p.headers[int(origin)-i]
+				if header == nil {
+					continue
+				}
+				headers = append(headers, header)
 			}
-			headers = append(headers, header)
 		}
 	}
 	atomic.AddUint64(&p.served, uint64(len(headers)))
@@ -705,6 +726,41 @@ func TestSkeletonSyncRetrievals(t *testing.T) {
 			endserve: (requestHeaders + 101 - 2) + (100 - 1), // midserve + lenrest - genesis
 			enddrop:  1,                                      // no new drops
 		},
+		// This test reproduces a bug caught during review (kudos to @holiman)
+		// where a subchain is merged with a previously interrupted one, causing
+		// pending data in the scratch space to become "invalid" (since we jump
+		// ahead during subchain merge). In that case it is expected to ignore
+		// the queued up data instead of trying to process on top of a shifted
+		// task set.
+		//
+		// The test is a bit convoluted since it needs to trigger a concurrency
+		// issue. First we sync up an initial chain of 2x512 items. Then announce
+		// 2x512+2 as head and delay delivering the head batch to fill the scratch
+		// space first. The delivery head should merge with the previous download
+		// and the scratch space must not be consumed further.
+		{
+			head: chain[2*requestHeaders],
+			peers: []*skeletonTestPeer{
+				newSkeletonTestPeerWithHook("peer-1", chain, func(origin uint64) []*types.Header {
+					if origin == chain[2*requestHeaders+2].Number.Uint64() {
+						time.Sleep(100 * time.Millisecond)
+					}
+					return nil // Fallback to default behavior, just delayed
+				}),
+				newSkeletonTestPeerWithHook("peer-2", chain, func(origin uint64) []*types.Header {
+					if origin == chain[2*requestHeaders+2].Number.Uint64() {
+						time.Sleep(100 * time.Millisecond)
+					}
+					return nil // Fallback to default behavior, just delayed
+				}),
+			},
+			midstate: []*subchain{{Head: 2 * requestHeaders, Tail: 1}},
+			midserve: 2*requestHeaders - 1, // len - head - genesis
+
+			newHead:  chain[2*requestHeaders+2],
+			endstate: []*subchain{{Head: 2*requestHeaders + 2, Tail: 1}},
+			endserve: 4 * requestHeaders,
+		},
 	}
 	for i, tt := range tests {
 		// Create a fresh database and initialize it with the starting state