Improved scanner modes part1 - refactor

cmd/es-node/config.go

@@ -85,6 +85,11 @@ func NewConfig(ctx *cli.Context, lg log.Logger) (*node.Config, error) {
 	// }
 
 	// l2SyncEndpoint := NewL2SyncEndpointConfig(ctx)
+
+	scannerConfig, err := scanner.NewConfig(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create scanner config: %w", err)
+	}
 	cfg := &node.Config{
 		L1:         *l1Endpoint,
 		ChainID:    chainId,
@@ -122,7 +127,7 @@ func NewConfig(ctx *cli.Context, lg log.Logger) (*node.Config, error) {
 		Storage:  *storageConfig,
 		Mining:   minerConfig,
 		Archiver: archiverConfig,
-		Scanner:  scanner.NewConfig(ctx),
+		Scanner:  scannerConfig,
 	}
 	if err := cfg.Check(); err != nil {
 		return nil, err

cmd/es-node/utils.go

@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"context"
-	"crypto/sha256"
 	"fmt"
 	"math/big"
 	"net/http"
@@ -19,11 +18,11 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/crypto/kzg4844"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rpc"
 	es "github.com/ethstorage/go-ethstorage/ethstorage"
+	"github.com/ethstorage/go-ethstorage/ethstorage/blobs"
 	"github.com/ethstorage/go-ethstorage/ethstorage/flags"
 	"github.com/ethstorage/go-ethstorage/ethstorage/storage"
 	"github.com/urfave/cli"
@@ -261,15 +260,11 @@ func downloadBlobFromRPC(endpoint string, kvIndex uint64, hash common.Hash) ([]b
 		return nil, err
 	}
 
-	var blob kzg4844.Blob
-	copy(blob[:], result)
-	commitment, err := kzg4844.BlobToCommitment(&blob)
+	blobhash, err := blobs.BlobToVersionedHash(result)
 	if err != nil {
 		return nil, fmt.Errorf("blobToCommitment failed: %w", err)
 	}
-	blobhash := common.Hash(kzg4844.CalcBlobHashV1(sha256.New(), &commitment))
-	fmt.Printf("blobhash from blob: %x\n", blobhash)
-	if bytes.Compare(blobhash[:es.HashSizeInContract], hash[:es.HashSizeInContract]) != 0 {
+	if !bytes.Equal(blobhash[:es.HashSizeInContract], hash[:es.HashSizeInContract]) {
 		return nil, fmt.Errorf("invalid blobhash for %d want: %x, got: %x", kvIndex, hash, blobhash)
 	}
 

ethstorage/node/node.go

@@ -341,6 +341,7 @@ func (n *EsNode) initScanner(ctx context.Context, cfg *Config) {
 		n.feed,
 		n.lg,
 	)
+	n.scanner.Start()
 }
 
 func (n *EsNode) Start(ctx context.Context, cfg *Config) error {

ethstorage/scanner/config.go

@@ -58,20 +58,20 @@ func CLIFlags() []cli.Flag {
 	return flags
 }
 
-func NewConfig(ctx *cli.Context) *Config {
+func NewConfig(ctx *cli.Context) (*Config, error) {
 	mode := ctx.GlobalInt(ModeFlagName)
 	if mode == modeDisabled {
-		return nil
+		return nil, nil
 	}
 	if mode != modeCheckMeta && mode != modeCheckBlob {
-		panic(fmt.Sprintf("invalid scanner mode: %d", mode))
+		return nil, fmt.Errorf("invalid scanner mode: %d", mode)
 	}
 	if interval := ctx.GlobalInt(IntervalFlagName); interval < defaultInterval {
-		panic(fmt.Sprintf("scanner interval must be at least %d minutes", defaultInterval))
+		return nil, fmt.Errorf("scanner interval must be at least %d minutes", defaultInterval)
 	}
 	return &Config{
 		Mode:      mode,
 		BatchSize: ctx.GlobalInt(BatchSizeFlagName),
 		Interval:  ctx.GlobalInt(IntervalFlagName),
-	}
+	}, nil
 }

ethstorage/scanner/scanner.go

@@ -22,9 +22,10 @@ type Scanner struct {
 	cancel    context.CancelFunc
 	wg        sync.WaitGroup
 	running   bool
-	mu        sync.Mutex
+	mu        sync.Mutex // protects running
 	lg        log.Logger
 	scanStats ScanStats
+	statsMu   sync.Mutex // protects scanStats
 }
 
 func New(
@@ -37,7 +38,7 @@ func New(
 	lg log.Logger,
 ) *Scanner {
 	cctx, cancel := context.WithCancel(ctx)
-	scanner := &Scanner{
+	return &Scanner{
 		worker:    NewWorker(sm, fetchBlob, l1, cfg, lg),
 		feed:      feed,
 		interval:  time.Minute * time.Duration(cfg.Interval),
@@ -46,9 +47,17 @@ func New(
 		lg:        lg,
 		scanStats: ScanStats{0, 0},
 	}
-	scanner.wg.Add(1)
-	go scanner.update()
-	return scanner
+}
+
+// Start begins the scanner's background processing. Must be called after New().
+func (s *Scanner) Start() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.running {
+		return
+	}
+	s.wg.Add(1)
+	go s.update()
 }
 
 func (s *Scanner) update() {
@@ -141,8 +150,11 @@ func (s *Scanner) logStats(sts *stats) {
 }
 
 func (s *Scanner) GetScanState() *ScanStats {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	if s == nil {
+		return &ScanStats{}
+	}
+	s.statsMu.Lock()
+	defer s.statsMu.Unlock()
 	snapshot := s.scanStats // Make a copy
 	return &snapshot        // Return a pointer to the copy
 }
@@ -161,11 +173,12 @@ func (s *Scanner) Close() {
 		return
 	}
 	s.running = false
+	cancel := s.cancel
 	s.mu.Unlock()
 
-	s.cancel()
-	s.lg.Info("Scanner closed")
+	cancel()
 	s.wg.Wait()
+	s.lg.Info("Scanner closed")
 }
 
 func (s *Scanner) doWork(tracker mismatchTracker) (*stats, scanErrors, error) {

ethstorage/scanner/utils.go

@@ -16,7 +16,7 @@ func (s scanErrors) add(kvIndex uint64, err error) {
 	s[kvIndex] = err
 }
 
-func (s scanErrors) nil(kvIndex uint64) {
+func (s scanErrors) clearError(kvIndex uint64) {
 	s[kvIndex] = nil
 }
 
@@ -162,25 +162,6 @@ func shortPrt(nums []uint64) string {
 	return strings.Join(res, ",")
 }
 
-func summaryLocalKvs(shards []uint64, kvEntries, lastKvIdx uint64) string {
-	var res []string
-	for _, shard := range shards {
-		if shard*kvEntries > lastKvIdx {
-			// skip empty shards
-			break
-		}
-		var lastEntry uint64
-		if shard == lastKvIdx/kvEntries {
-			lastEntry = lastKvIdx
-		} else {
-			lastEntry = (shard+1)*kvEntries - 1
-		}
-		shardView := fmt.Sprintf("shard%d%s", shard, formatRange(shard*kvEntries, lastEntry))
-		res = append(res, shardView)
-	}
-	return strings.Join(res, ",")
-}
-
 func formatRange(start, end uint64) string {
 	if start == end {
 		return fmt.Sprintf("[%d]", start)