fix(headers): signal cleanly shutdown headers and allow exit (#5286)

eth/stagedsync/stage_headers.go

@@ -3,7 +3,6 @@ package stagedsync
 import (
 	"context"
 	"encoding/binary"
-	"errors"
 	"fmt"
 	"math/big"
 	"runtime"
@@ -706,7 +705,8 @@ func forkingPoint(
 func handleInterrupt(interrupt engineapi.Interrupt, cfg HeadersCfg, tx kv.RwTx, headerInserter *headerdownload.HeaderInserter, useExternalTx bool) (bool, error) {
 	if interrupt != engineapi.None {
 		if interrupt == engineapi.Stopping {
-			cfg.hd.PayloadStatusCh <- engineapi.PayloadStatus{CriticalError: errors.New("server is stopping")}
+			close(cfg.hd.ShutdownCh)
+			return false, fmt.Errorf("server is stopping")
 		}
 		if interrupt == engineapi.Synced && cfg.hd.HeadersCollector() != nil {
 			verifyAndSaveDownloadedPoSHeaders(tx, cfg, headerInserter)

turbo/stages/headerdownload/header_data_struct.go

@@ -309,6 +309,7 @@ type HeaderDownload struct {
 	headersCollector     *etl.Collector               // ETL collector for headers
 	BeaconRequestList    *engineapi.RequestList       // Requests from ethbackend to staged sync
 	PayloadStatusCh      chan engineapi.PayloadStatus // Responses (validation/execution status)
+	ShutdownCh           chan struct{}                // Channel to signal shutdown
 	pendingPayloadHash   common.Hash                  // Header whose status we still should send to PayloadStatusCh
 	pendingPayloadStatus *engineapi.PayloadStatus     // Alternatively, there can be an already prepared response to send to PayloadStatusCh
 	unsettledForkChoice  *engineapi.ForkChoiceMessage // Forkchoice to process after unwind
@@ -344,6 +345,7 @@ func NewHeaderDownload(
 		QuitPoWMining:      make(chan struct{}),
 		BeaconRequestList:  engineapi.NewRequestList(),
 		PayloadStatusCh:    make(chan engineapi.PayloadStatus, 1),
+		ShutdownCh:         make(chan struct{}),
 		headerReader:       headerReader,
 		badPoSHeaders:      make(map[common.Hash]common.Hash),
 	}

turbo/stages/stageloop.go

@@ -79,6 +79,13 @@ func StageLoop(
 	for {
 		start := time.Now()
 
+		select {
+		case <-hd.ShutdownCh:
+			return
+		default:
+			// continue
+		}
+
 		// Estimate the current top height seen from the peer
 		height := hd.TopSeenHeight()
 		headBlockHash, err := StageLoopStep(ctx, db, sync, height, notifications, initialCycle, updateHead, nil)