Added blob sidecars snapshots for sepolia

cl/phase1/stages/stage_history_download.go

@@ -91,7 +91,6 @@ func SpawnStageHistoryDownload(cfg StageHistoryReconstructionCfg, ctx context.Co
 
 	var currEth1Progress atomic.Int64
 
-	bytesReadInTotal := atomic.Uint64{}
 	destinationSlotForEL := uint64(math.MaxUint64)
 	if cfg.engine != nil && cfg.engine.SupportInsertion() && cfg.beaconCfg.DenebForkEpoch != math.MaxUint64 {
 		destinationSlotForEL = cfg.beaconCfg.BellatrixForkEpoch * cfg.beaconCfg.SlotsPerEpoch
@@ -109,8 +108,6 @@ func SpawnStageHistoryDownload(cfg StageHistoryReconstructionCfg, ctx context.Co
 
 		destinationSlotForCL := cfg.sn.SegmentsMax()
 
-		bytesReadInTotal.Add(uint64(blk.EncodingSizeSSZ()))
-
 		slot := blk.Block.Slot
 		if destinationSlotForCL <= blk.Block.Slot {
 			if err := beacon_indicies.WriteBeaconBlockAndIndicies(ctx, tx, blk, true); err != nil {
@@ -154,6 +151,8 @@ func SpawnStageHistoryDownload(cfg StageHistoryReconstructionCfg, ctx context.Co
 	finishCh := make(chan struct{})
 	// Start logging thread
 
+	isBackfilling := atomic.Bool{}
+
 	go func() {
 		logInterval := time.NewTicker(logIntervalTime)
 		defer logInterval.Stop()
@@ -177,24 +176,21 @@ func SpawnStageHistoryDownload(cfg StageHistoryReconstructionCfg, ctx context.Co
 				ratio := float64(logTime / time.Second)
 				speed := blockProgress / ratio
 				prevProgress = currProgress
-				peerCount, err := cfg.downloader.Peers()
-				if err != nil {
-					log.Debug("could not get peer count", "err", err)
-					continue
-				}
+
 				if speed == 0 {
 					continue
 				}
 				logArgs = append(logArgs,
 					"slot", currProgress,
 					"blockNumber", currEth1Progress.Load(),
 					"blk/sec", fmt.Sprintf("%.1f", speed),
-					"mbps/sec", fmt.Sprintf("%.1f", float64(bytesReadInTotal.Load())/(1000*1000*ratio)),
-					"peers", peerCount,
 					"snapshots", cfg.sn.SegmentsMax(),
 				)
-				bytesReadInTotal.Store(0)
-				logger.Info("Backfilling History", logArgs...)
+				logMsg := "Node is still syncing... downloading past blocks"
+				if isBackfilling.Load() {
+					logMsg = "Node has finished syncing... full history is being downloaded for archiving purposes"
+				}
+				logger.Info(logMsg, logArgs...)
 			case <-finishCh:
 				return
 			case <-ctx.Done():
@@ -211,7 +207,11 @@ func SpawnStageHistoryDownload(cfg StageHistoryReconstructionCfg, ctx context.Co
 			}
 		}
 		cfg.antiquary.NotifyBackfilled()
-		log.Info("Backfilling finished")
+		if cfg.backfilling {
+			cfg.logger.Info("full backfilling finished")
+		} else {
+			cfg.logger.Info("Missing blocks download finished (note: this does not mean that the history is complete, only that the missing blocks need for sync have been downloaded)")
+		}
 
 		close(finishCh)
 		if cfg.blobsBackfilling {
@@ -240,6 +240,7 @@ func SpawnStageHistoryDownload(cfg StageHistoryReconstructionCfg, ctx context.Co
 		return err
 	}
 	defer tx2.Rollback()
+	isBackfilling.Store(true)
 
 	cfg.logger.Info("Ready to insert history, waiting for sync cycle to finish")
 

erigon-lib/downloader/downloader.go

@@ -139,7 +139,6 @@ func New(ctx context.Context, cfg *downloadercfg.Cfg, dirs datadir.Dirs, logger
 	var stats AggStats
 
 	lock, err := getSnapshotLock(ctx, cfg, db, &stats, mutex, logger)
-
 	if err != nil {
 		return nil, fmt.Errorf("can't initialize snapshot lock: %w", err)
 	}
@@ -658,7 +657,7 @@ func (d *Downloader) mainLoop(silent bool) error {
 		go func() {
 			defer d.wg.Done()
 			d.webseeds.Discover(d.ctx, d.cfg.WebSeedUrls, d.cfg.WebSeedFiles, d.cfg.Dirs.Snap)
-			// webseeds.Discover may create new .torrent files on disk
+			// apply webseeds to existing torrents
 			if err := d.addTorrentFilesFromDisk(true); err != nil && !errors.Is(err, context.Canceled) {
 				d.logger.Warn("[snapshots] addTorrentFilesFromDisk", "err", err)
 			}
@@ -2140,8 +2139,12 @@ func (d *Downloader) AddMagnetLink(ctx context.Context, infoHash metainfo.Hash,
 	if d.alreadyHaveThisName(name) || !IsSnapNameAllowed(name) {
 		return nil
 	}
+	isProhibited, err := d.torrentFiles.newDownloadsAreProhibited(name)
+	if err != nil {
+		return err
+	}
 
-	if d.torrentFiles.newDownloadsAreProhibited() && !d.torrentFiles.Exists(name) {
+	if isProhibited && !d.torrentFiles.Exists(name) {
 		return nil
 	}
 
@@ -2168,8 +2171,33 @@ func (d *Downloader) AddMagnetLink(ctx context.Context, infoHash metainfo.Hash,
 		case <-ctx.Done():
 			return
 		case <-t.GotInfo():
+		case <-time.After(30 * time.Second): //fallback to r2
+			// TOOD: handle errors
+			// TOOD: add `d.webseeds.Complete` chan - to prevent race - Discover is also async
+			// TOOD: maybe run it in goroutine and return channel - to select with p2p
+
+			ok, err := d.webseeds.DownloadAndSaveTorrentFile(ctx, name)
+			if ok && err == nil {
+				ts, err := d.torrentFiles.LoadByPath(filepath.Join(d.SnapDir(), name+".torrent"))
+				if err != nil {
+					return
+				}
+				_, _, err = addTorrentFile(ctx, ts, d.torrentClient, d.db, d.webseeds)
+				if err != nil {
+					return
+				}
+				return
+			}
+
+			// wait for p2p
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.GotInfo():
+			}
 		}
 
+		//TODO: remove whitelist check - Erigon may send us new seedable files
 		if !d.snapshotLock.Downloads.Contains(name) {
 			mi := t.Metainfo()
 			if err := CreateTorrentFileIfNotExists(d.SnapDir(), t.Info(), &mi, d.torrentFiles); err != nil {

erigon-lib/downloader/downloader_grpc_server.go

@@ -45,11 +45,8 @@ type GrpcServer struct {
 	d *Downloader
 }
 
-func (s *GrpcServer) ProhibitNewDownloads(context.Context, *proto_downloader.ProhibitNewDownloadsRequest) (*emptypb.Empty, error) {
-	if err := s.d.torrentFiles.prohibitNewDownloads(); err != nil {
-		return nil, err
-	}
-	return nil, nil
+func (s *GrpcServer) ProhibitNewDownloads(ctx context.Context, req *proto_downloader.ProhibitNewDownloadsRequest) (*emptypb.Empty, error) {
+	return &emptypb.Empty{}, s.d.torrentFiles.prohibitNewDownloads(req.Type)
 }
 
 // Erigon "download once" - means restart/upgrade/downgrade will not download files (and will be fast)

erigon-lib/downloader/snaptype/type.go

@@ -147,7 +147,6 @@ func (s snapType) IdxFileNames(version Version, from uint64, to uint64) []string
 }
 
 func (s snapType) IdxFileName(version Version, from uint64, to uint64, index ...Index) string {
-
 	if len(index) == 0 {
 		if len(s.indexes) == 0 {
 			return ""
@@ -344,7 +343,7 @@ var (
 
 	BorSnapshotTypes = []Type{BorEvents, BorSpans}
 
-	CaplinSnapshotTypes = []Type{BeaconBlocks}
+	CaplinSnapshotTypes = []Type{BeaconBlocks, BlobSidecars}
 
 	AllTypes = []Type{
 		Headers,
@@ -353,5 +352,6 @@ var (
 		BorEvents,
 		BorSpans,
 		BeaconBlocks,
+		BlobSidecars,
 	}
 )

erigon-lib/downloader/torrent_files.go

@@ -1,7 +1,9 @@
 package downloader
 
 import (
+	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -10,6 +12,7 @@ import (
 	"github.com/anacrolix/torrent"
 	"github.com/anacrolix/torrent/metainfo"
 	"github.com/ledgerwatch/erigon-lib/common/dir"
+	"golang.org/x/exp/slices"
 )
 
 // TorrentFiles - does provide thread-safe CRUD operations on .torrent files
@@ -125,29 +128,70 @@ const ProhibitNewDownloadsFileName = "prohibit_new_downloads.lock"
 // Erigon "download once" - means restart/upgrade/downgrade will not download files (and will be fast)
 // After "download once" - Erigon will produce and seed new files
 // Downloader will able: seed new files (already existing on FS), download uncomplete parts of existing files (if Verify found some bad parts)
-func (tf *TorrentFiles) prohibitNewDownloads() error {
+func (tf *TorrentFiles) prohibitNewDownloads(t string) error {
 	tf.lock.Lock()
 	defer tf.lock.Unlock()
-	return CreateProhibitNewDownloadsFile(tf.dir)
-}
-func (tf *TorrentFiles) newDownloadsAreProhibited() bool {
-	tf.lock.Lock()
-	defer tf.lock.Unlock()
-	return dir.FileExist(filepath.Join(tf.dir, ProhibitNewDownloadsFileName))
+	// open or create file ProhibitNewDownloadsFileName
+	f, err := os.OpenFile(filepath.Join(tf.dir, ProhibitNewDownloadsFileName), os.O_CREATE|os.O_RDONLY, 0644)
+	if err != nil {
+		return fmt.Errorf("open file: %w", err)
+	}
+	defer f.Close()
+	var prohibitedList []string
+	torrentListJsonBytes, err := io.ReadAll(f)
+	if err != nil {
+		return fmt.Errorf("read file: %w", err)
+	}
+	if len(torrentListJsonBytes) > 0 {
+		if err := json.Unmarshal(torrentListJsonBytes, &prohibitedList); err != nil {
+			return fmt.Errorf("unmarshal: %w", err)
+		}
+	}
+	if slices.Contains(prohibitedList, t) {
+		return nil
+	}
+	prohibitedList = append(prohibitedList, t)
+	f.Close()
 
-	//return dir.FileExist(filepath.Join(tf.dir, ProhibitNewDownloadsFileName)) ||
-	//	dir.FileExist(filepath.Join(tf.dir, SnapshotsLockFileName))
+	// write new prohibited list by opening the file in truncate mode
+	f, err = os.OpenFile(filepath.Join(tf.dir, ProhibitNewDownloadsFileName), os.O_TRUNC|os.O_WRONLY, 0644)
+	if err != nil {
+		return fmt.Errorf("open file for writing: %w", err)
+	}
+	defer f.Close()
+	prohibitedListJsonBytes, err := json.Marshal(prohibitedList)
+	if err != nil {
+		return fmt.Errorf("marshal: %w", err)
+	}
+	if _, err := f.Write(prohibitedListJsonBytes); err != nil {
+		return fmt.Errorf("write: %w", err)
+	}
+
+	return f.Sync()
 }
 
-func CreateProhibitNewDownloadsFile(dir string) error {
-	fPath := filepath.Join(dir, ProhibitNewDownloadsFileName)
-	f, err := os.Create(fPath)
+func (tf *TorrentFiles) newDownloadsAreProhibited(name string) (bool, error) {
+	tf.lock.Lock()
+	defer tf.lock.Unlock()
+	f, err := os.OpenFile(filepath.Join(tf.dir, ProhibitNewDownloadsFileName), os.O_CREATE|os.O_APPEND|os.O_RDONLY, 0644)
 	if err != nil {
-		return err
+		return false, err
 	}
 	defer f.Close()
-	if err := f.Sync(); err != nil {
-		return err
+	var prohibitedList []string
+	torrentListJsonBytes, err := io.ReadAll(f)
+	if err != nil {
+		return false, fmt.Errorf("newDownloadsAreProhibited: read file: %w", err)
 	}
-	return nil
+	if len(torrentListJsonBytes) > 0 {
+		if err := json.Unmarshal(torrentListJsonBytes, &prohibitedList); err != nil {
+			return false, fmt.Errorf("newDownloadsAreProhibited: unmarshal: %w", err)
+		}
+	}
+	for _, p := range prohibitedList {
+		if strings.Contains(name, p) {
+			return true, nil
+		}
+	}
+	return false, nil
 }

erigon-lib/downloader/util.go

@@ -340,6 +340,7 @@ func _addTorrentFile(ctx context.Context, ts *torrent.TorrentSpec, torrentClient
 				return nil, false, fmt.Errorf("addTorrentFile %s: update failed: %w", ts.DisplayName, err)
 			}
 		} else {
+			t.AddWebSeeds(ts.Webseeds)
 			if err := db.Update(ctx, torrentInfoReset(ts.DisplayName, ts.InfoHash.Bytes(), 0)); err != nil {
 				return nil, false, fmt.Errorf("addTorrentFile %s: reset failed: %w", ts.DisplayName, err)
 			}

erigon-lib/downloader/webseed.go

@@ -40,9 +40,6 @@ type WebSeeds struct {
 }
 
 func (d *WebSeeds) Discover(ctx context.Context, urls []*url.URL, files []string, rootDir string) {
-	if d.torrentFiles.newDownloadsAreProhibited() {
-		return
-	}
 	listsOfFiles := d.constructListsOfFiles(ctx, urls, files)
 	torrentMap := d.makeTorrentUrls(listsOfFiles)
 	webSeedMap := d.downloadTorrentFilesFromProviders(ctx, rootDir, torrentMap)
@@ -64,6 +61,14 @@ func (d *WebSeeds) constructListsOfFiles(ctx context.Context, httpProviders []*u
 			d.logger.Debug("[snapshots.webseed] get from HTTP provider", "err", err, "url", webSeedProviderURL.EscapedPath())
 			continue
 		}
+		// check if we need to prohibit new downloads for some files
+		for name := range manifestResponse {
+			prohibited, err := d.torrentFiles.newDownloadsAreProhibited(name)
+			if prohibited || err != nil {
+				delete(manifestResponse, name)
+			}
+		}
+
 		listsOfFiles = append(listsOfFiles, manifestResponse)
 	}
 
@@ -74,6 +79,13 @@ func (d *WebSeeds) constructListsOfFiles(ctx context.Context, httpProviders []*u
 			d.logger.Debug("[snapshots.webseed] get from File provider", "err", err)
 			continue
 		}
+		// check if we need to prohibit new downloads for some files
+		for name := range response {
+			prohibited, err := d.torrentFiles.newDownloadsAreProhibited(name)
+			if prohibited || err != nil {
+				delete(response, name)
+			}
+		}
 		listsOfFiles = append(listsOfFiles, response)
 	}
 	return listsOfFiles
@@ -219,17 +231,13 @@ func (d *WebSeeds) downloadTorrentFilesFromProviders(ctx context.Context, rootDi
 		tUrls := tUrls
 		e.Go(func() error {
 			for _, url := range tUrls {
-				res, err := d.callTorrentHttpProvider(ctx, url, name)
+				//validation happens inside
+				_, err := d.callTorrentHttpProvider(ctx, url, name)
 				if err != nil {
 					d.logger.Log(d.verbosity, "[snapshots] got from webseed", "name", name, "err", err, "url", url)
 					continue
 				}
-				if !d.torrentFiles.Exists(name) {
-					if err := d.torrentFiles.Create(name, res); err != nil {
-						d.logger.Log(d.verbosity, "[snapshots] .torrent from webseed rejected", "name", name, "err", err, "url", url)
-						continue
-					}
-				}
+				//don't save .torrent here - do it inside downloader.Add
 				webSeeMapLock.Lock()
 				webSeedMap[torrentMap[*url]] = struct{}{}
 				webSeeMapLock.Unlock()
@@ -244,6 +252,33 @@ func (d *WebSeeds) downloadTorrentFilesFromProviders(ctx context.Context, rootDi
 	return webSeedMap
 }
 
+func (d *WebSeeds) DownloadAndSaveTorrentFile(ctx context.Context, name string) (bool, error) {
+	urls, ok := d.ByFileName(name)
+	if !ok {
+		return false, nil
+	}
+	for _, urlStr := range urls {
+		parsedUrl, err := url.Parse(urlStr)
+		if err != nil {
+			continue
+		}
+		res, err := d.callTorrentHttpProvider(ctx, parsedUrl, name)
+		if err != nil {
+			return false, err
+		}
+		if d.torrentFiles.Exists(name) {
+			continue
+		}
+		if err := d.torrentFiles.Create(name, res); err != nil {
+			d.logger.Log(d.verbosity, "[snapshots] .torrent from webseed rejected", "name", name, "err", err)
+			continue
+		}
+		return true, nil
+	}
+
+	return false, nil
+}
+
 func (d *WebSeeds) callTorrentHttpProvider(ctx context.Context, url *url.URL, fileName string) ([]byte, error) {
 	request, err := http.NewRequest(http.MethodGet, url.String(), nil)
 	if err != nil {

erigon-lib/go.mod

@@ -4,7 +4,7 @@ go 1.20
 
 require (
 	github.com/erigontech/mdbx-go v0.27.24
-	github.com/ledgerwatch/erigon-snapshot v1.3.1-0.20240222083139-3cef6c872d07
+	github.com/ledgerwatch/erigon-snapshot v1.3.1-0.20240321134048-58ba2110522a
 	github.com/ledgerwatch/interfaces v0.0.0-20240320062914-b57f05746087
 	github.com/ledgerwatch/log/v3 v3.9.0
 	github.com/ledgerwatch/secp256k1 v1.0.0