Snapshot: add hash first byte to headers.seg, serve p2p blocks from s…

…napshots (erigontech#3198)

cmd/abigen/main.go

@@ -115,7 +115,6 @@ func init() {
 		aliasFlag,
 	}
 	app.Action = abigen
-	cli.CommandHelpTemplate = flags.OriginCommandHelpTemplate
 }
 
 func abigen(c *cli.Context) error {

cmd/downloader/downloader/downloader.go

@@ -27,15 +27,16 @@ type Client struct {
 	pieceCompletionStore storage.PieceCompletion
 }
 
-func TorrentConfig(snapshotsDir string, seeding bool, peerID string, verbosity lg.Level, downloadLimit, uploadLimit datasize.ByteSize) (*torrent.ClientConfig, storage.PieceCompletion, error) {
+func TorrentConfig(snapshotsDir string, seeding bool, peerID string, verbosity lg.Level, downloadRate, uploadRate datasize.ByteSize) (*torrent.ClientConfig, storage.PieceCompletion, error) {
 	torrentConfig := DefaultTorrentConfig()
 	torrentConfig.Seed = seeding
 	torrentConfig.DataDir = snapshotsDir
 	torrentConfig.UpnpID = torrentConfig.UpnpID + "leecher"
 	torrentConfig.PeerID = peerID
 
-	torrentConfig.UploadRateLimiter = rate.NewLimiter(rate.Limit(downloadLimit.Bytes()), 2*DefaultPieceSize) // default: unlimited
-	torrentConfig.DownloadRateLimiter = rate.NewLimiter(rate.Limit(uploadLimit.Bytes()), 2*DefaultPieceSize) // default: unlimited
+	// rates are divided by 2 - I don't know why it works, maybe bug inside torrent lib accounting
+	torrentConfig.UploadRateLimiter = rate.NewLimiter(rate.Limit(uploadRate.Bytes()/2), 2*DefaultPieceSize)     // default: unlimited
+	torrentConfig.DownloadRateLimiter = rate.NewLimiter(rate.Limit(downloadRate.Bytes()/2), 2*DefaultPieceSize) // default: unlimited
 
 	// debug
 	if lg.Debug == verbosity {

cmd/downloader/main.go

@@ -36,12 +36,13 @@ import (
 )
 
 var (
-	datadir                          string
-	seeding                          bool
-	asJson                           bool
-	downloaderApiAddr                string
-	torrentVerbosity                 string
-	downloadLimitStr, uploadLimitStr string
+	datadir                       string
+	seeding                       bool
+	asJson                        bool
+	forceRebuild                  bool
+	downloaderApiAddr             string
+	torrentVerbosity              string
+	downloadRateStr, uploadRteStr string
 )
 
 func init() {
@@ -53,11 +54,13 @@ func init() {
 	rootCmd.PersistentFlags().BoolVar(&seeding, "seeding", true, "Seed snapshots")
 	rootCmd.Flags().StringVar(&downloaderApiAddr, "downloader.api.addr", "127.0.0.1:9093", "external downloader api network address, for example: 127.0.0.1:9093 serves remote downloader interface")
 	rootCmd.Flags().StringVar(&torrentVerbosity, "torrent.verbosity", lg.Info.LogString(), "DEBUG | INFO | WARN | ERROR")
-	rootCmd.Flags().StringVar(&downloadLimitStr, "download.limit", "1gb", "bytes per second, example: 32mb")
-	rootCmd.Flags().StringVar(&uploadLimitStr, "upload.limit", "1gb", "bytes per second, example: 32mb")
+	rootCmd.Flags().StringVar(&downloadRateStr, "download.rate", "1gb", "bytes per second, example: 32mb")
+	rootCmd.Flags().StringVar(&uploadRteStr, "upload.rate", "1gb", "bytes per second, example: 32mb")
 
 	withDatadir(printInfoHashes)
 	printInfoHashes.PersistentFlags().BoolVar(&asJson, "json", false, "Print in json format (default: toml)")
+	printInfoHashes.PersistentFlags().BoolVar(&forceRebuild, "rebuild", false, "Force re-create .torrent files")
+
 	rootCmd.AddCommand(printInfoHashes)
 }
 
@@ -106,15 +109,15 @@ func Downloader(ctx context.Context, cmd *cobra.Command) error {
 		panic(fmt.Errorf("unexpected torrent.verbosity level: %s", torrentVerbosity))
 	}
 
-	var downloadLimit, uploadLimit datasize.ByteSize
-	if err := downloadLimit.UnmarshalText([]byte(downloadLimitStr)); err != nil {
+	var downloadRate, uploadRate datasize.ByteSize
+	if err := downloadRate.UnmarshalText([]byte(downloadRateStr)); err != nil {
 		return err
 	}
-	if err := uploadLimit.UnmarshalText([]byte(uploadLimitStr)); err != nil {
+	if err := uploadRate.UnmarshalText([]byte(uploadRteStr)); err != nil {
 		return err
 	}
 
-	log.Info("Run snapshot downloader", "addr", downloaderApiAddr, "datadir", datadir, "seeding", seeding)
+	log.Info("Run snapshot downloader", "addr", downloaderApiAddr, "datadir", datadir, "seeding", seeding, "download.rate", downloadRate.String(), "upload.rate", uploadRate.String())
 	if err := os.MkdirAll(snapshotsDir, 0755); err != nil {
 		return err
 	}
@@ -127,7 +130,7 @@ func Downloader(ctx context.Context, cmd *cobra.Command) error {
 			return fmt.Errorf("get peer id: %w", err)
 		}
 
-		cfg, pieceStore, err := downloader.TorrentConfig(snapshotsDir, seeding, string(peerID), torrentLogLevel, downloadLimit, uploadLimit)
+		cfg, pieceStore, err := downloader.TorrentConfig(snapshotsDir, seeding, string(peerID), torrentLogLevel, downloadRate, uploadRate)
 		if err != nil {
 			return err
 		}
@@ -195,10 +198,22 @@ func Downloader(ctx context.Context, cmd *cobra.Command) error {
 }
 
 var printInfoHashes = &cobra.Command{
-	Use:     "print_torrent_files",
-	Example: "go run ./cmd/downloader print_info_hashes --datadir <your_datadir> ",
+	Use:     "info_hashes",
+	Example: "go run ./cmd/downloader info_hashes --datadir <your_datadir>",
 	RunE: func(cmd *cobra.Command, args []string) error {
 		snapshotsDir := path.Join(datadir, "snapshots")
+		ctx := cmd.Context()
+
+		if forceRebuild { // remove and create .torrent files (will re-read all snapshots)
+			if err := downloader.ForEachTorrentFile(snapshotsDir, func(torrentFilePath string) error {
+				return os.Remove(torrentFilePath)
+			}); err != nil {
+				return err
+			}
+			if err := downloader.BuildTorrentFilesIfNeed(ctx, snapshotsDir); err != nil {
+				return err
+			}
+		}
 
 		res := map[string]string{}
 		err := downloader.ForEachTorrentFile(snapshotsDir, func(torrentFilePath string) error {

cmd/downloader/readme.md

@@ -41,10 +41,18 @@ erigon --downloader.api.addr=127.0.0.1:9093 --experimental.snapshot
 downloader --download.limit=10mb --upload.limit=10mb
 ```
 
-### Add hashes to https://github.com/ledgerwatch/erigon-snapshot
+### Print info_hashes in format compatible with https://github.com/ledgerwatch/erigon-snapshot
 
 ```
-downloader print_torrent_files --datadir=<your_datadir>
+downloader info_hashes --datadir=<your_datadir>
+```
+
+### Force re-calculate info_hashes and print in format compatible with https://github.com/ledgerwatch/erigon-snapshot
+
+```
+// will re-read all .seg files (high disk IO)
+// also does remove and create .torrent files
+downloader info_hashes --datadir=<your_datadir> --rebuild
 ```
 
 ### Create new snapshots
@@ -61,3 +69,6 @@ rsync server1:<your_datadir>/snapshots/*.torrent server2:<your_datadir>/snapshot
 // re-start downloader 
 ```
 
+## Known Issues
+
+- RPCDaemon with --datadir option need restart to make new segments available

cmd/evm/main.go

@@ -186,7 +186,6 @@ func init() {
 		stateTestCommand,
 		stateTransitionCommand,
 	}
-	cli.CommandHelpTemplate = flags.OriginCommandHelpTemplate
 }
 
 func main() {

cmd/rpcdaemon/cli/config.go

@@ -279,7 +279,10 @@ func RemoteServices(ctx context.Context, cfg Flags, logger log.Logger, rootCance
 			}
 
 			allSnapshots := snapshotsync.NewAllSnapshots(cfg.Snapshot.Dir, snapshothashes.KnownConfig(cc.ChainName))
-			if err != nil {
+			if err := allSnapshots.ReopenSegments(); err != nil {
+				return nil, nil, nil, nil, nil, nil, err
+			}
+			if err := allSnapshots.ReopenIndices(); err != nil {
 				return nil, nil, nil, nil, nil, nil, err
 			}
 			blockReader = snapshotsync.NewBlockReaderWithSnapshots(allSnapshots)

cmd/utils/flags.go

@@ -45,7 +45,6 @@ import (
 	"github.com/ledgerwatch/erigon/crypto"
 	"github.com/ledgerwatch/erigon/eth/ethconfig"
 	"github.com/ledgerwatch/erigon/eth/gasprice"
-	"github.com/ledgerwatch/erigon/internal/flags"
 	"github.com/ledgerwatch/erigon/metrics"
 	"github.com/ledgerwatch/erigon/node"
 	"github.com/ledgerwatch/erigon/p2p"
@@ -69,16 +68,14 @@ GLOBAL OPTIONS:
    {{range .Flags}}{{.}}
    {{end}}{{end}}
 `
-	cli.CommandHelpTemplate = flags.CommandHelpTemplate
 	cli.HelpPrinter = printHelp
 }
 
 func printHelp(out io.Writer, templ string, data interface{}) {
 	funcMap := template.FuncMap{"join": strings.Join}
 	t := template.Must(template.New("help").Funcs(funcMap).Parse(templ))
 	w := tabwriter.NewWriter(out, 38, 8, 2, ' ', 0)
-	err := t.Execute(w, data)
-	if err != nil {
+	if err := t.Execute(w, data); err != nil {
 		panic(err)
 	}
 	w.Flush()

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/VictoriaMetrics/metrics v1.18.1
 	github.com/anacrolix/go-libutp v1.1.0
 	github.com/anacrolix/log v0.10.0
-	github.com/anacrolix/torrent v1.39.1
+	github.com/anacrolix/torrent v1.40.0
 	github.com/btcsuite/btcd v0.21.0-beta
 	github.com/c2h5oh/datasize v0.0.0-20200825124411-48ed595a09d2
 	github.com/consensys/gnark-crypto v0.4.0
@@ -21,7 +21,7 @@ require (
 	github.com/emicklei/dot v0.16.0
 	github.com/fatih/color v1.12.0
 	github.com/fjl/gencodec v0.0.0-20191126094850-e283372f291f
-	github.com/flanglet/kanzi-go v1.9.0
+	github.com/flanglet/kanzi-go v1.9.1-0.20211212184056-72dda96261ee
 	github.com/go-sourcemap/sourcemap v2.1.3+incompatible // indirect
 	github.com/goccy/go-json v0.7.4
 	github.com/gofrs/flock v0.8.1
@@ -45,8 +45,8 @@ require (
 	github.com/pelletier/go-toml v1.9.3
 	github.com/pelletier/go-toml/v2 v2.0.0-beta.4
 	github.com/quasilyte/go-ruleguard/dsl v0.3.6
-	github.com/rs/cors v1.8.0
-	github.com/shirou/gopsutil/v3 v3.21.11
+	github.com/rs/cors v1.8.2
+	github.com/shirou/gopsutil/v3 v3.21.12
 	github.com/spf13/cobra v1.2.1
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.7.1-0.20210427113832-6241f9ab9942