古い署名を削除 ＆ Streamで同期しない

Makefile

@@ -1,8 +1,16 @@
 build:
 	go build -o bin/oasvlfy .
 
+.PHONY: proto
+proto:
+	protoc --go_out=./proto/p2p/v1  \
+	./proto/p2p/v1/*.proto
+
 fmt:
 	go fmt ./...
 
+fmtproto:
+	clang-format -i ./proto/p2p/**/*.proto
+
 test:
 	go test -v ./...

cmd/config_loader.go

@@ -1,7 +1,7 @@
 package cmd
 
 import (
-	"io/ioutil"
+	"os"
 	"strings"
 
 	"github.com/oasysgames/oasys-optimism-verifier/config"
@@ -139,20 +139,20 @@ func mustNewConfigLoader(cmd *cobra.Command) *configLoader {
 	return opts
 }
 
-func (opts *configLoader) load() (*config.Config, error) {
+func (opts *configLoader) load(enableStrictValidation bool) (*config.Config, error) {
 	// load config from the file
 	if !opts.fromCli {
 		path, err := opts.cmd.Flags().GetString(fileConfigFlag)
 		if err != nil {
 			return nil, err
 		}
 
-		input, err := ioutil.ReadFile(path)
+		input, err := os.ReadFile(path)
 		if err != nil {
 			return nil, err
 		}
 
-		conf, err := config.NewConfig(input)
+		conf, err := config.NewConfig(input, true)
 		if err != nil {
 			return nil, err
 		}
@@ -188,7 +188,7 @@ func (opts *configLoader) load() (*config.Config, error) {
 		}
 	}
 
-	if err := config.Validate(opts.cfg); err != nil {
+	if err := config.Validate(opts.cfg, enableStrictValidation); err != nil {
 		return nil, err
 	}
 

cmd/config_loader_test.go

@@ -57,11 +57,11 @@ func (s *ConfigLoaderTestSuite) TestLoadConfigFromYAML() {
 			address: '0xD244F03CA3e99C6093f6cBEFBD2f4508244C59D4'
 			password: %s
 			plain: '0xebf3a7f5f805e02c0bbbd599acd5c881f40db22caa95127d4bf48e2dde5fd7bb'
-	
+
 	hub_layer:
 		chain_id: 1
 		rpc: https://rpc.hub.example.com/
-	
+
 	verse_layer:
 		discovery:
 			endpoint: https://discovery.example.com/
@@ -71,7 +71,7 @@ func (s *ConfigLoaderTestSuite) TestLoadConfigFromYAML() {
 			  l1_contracts:
 			    StateCommitmentChain: '0x01E901F3c65fA7CBd4505F5eF3A88e4ce432e4B5'
 			    L2OutputOracle: '0x2489317FA6e003550111D5D196302Ba0879354e2'
-	
+
 	p2p:
 		listens:
 			- listen0
@@ -92,7 +92,7 @@ func (s *ConfigLoaderTestSuite) TestLoadConfigFromYAML() {
 	verifier:
 		enable: true
 		wallet: verifier
-	
+
 	submitter:
 		enable: true
 		confirmations: 10
@@ -123,7 +123,7 @@ func (s *ConfigLoaderTestSuite) TestLoadConfigFromYAML() {
 	})
 	rootCmd.Execute()
 
-	got, _ := globalConfigLoader.load()
+	got, _ := globalConfigLoader.load(false)
 	s.Equal(want, got)
 }
 
@@ -227,7 +227,8 @@ func (s *ConfigLoaderTestSuite) executeWithCliArgs(appendArgs []string) *config.
 	}, appendArgs...))
 	cmd.Execute()
 
-	conf, _ := opts.load()
+	conf, err := opts.load(false)
+	s.Require().NoError(err)
 	return conf
 }
 
@@ -336,6 +337,8 @@ func (s *ConfigLoaderTestSuite) configWithMinCliArgs() *config.Config {
 			StateCollectLimit:   defaults["verifier.state_collect_limit"].(int),
 			StateCollectTimeout: defaults["verifier.state_collect_timeout"].(time.Duration),
 			OptimizeInterval:    defaults["verifier.db_optimize_interval"].(time.Duration),
+			Confirmations:       defaults["verifier.confirmations"].(int),
+			StartBlockOffset:    defaults["verifier.start_block_offset"].(uint64),
 		},
 		Submitter: config.Submitter{
 			Enable:              false,

cmd/ping.go

@@ -16,7 +16,7 @@ var pingCmd = &cobra.Command{
 	Short: "Send ping via P2P to specified peer",
 	Long:  "Send ping via P2P to specified peer",
 	Run: func(cmd *cobra.Command, args []string) {
-		conf, err := globalConfigLoader.load()
+		conf, err := globalConfigLoader.load(true)
 		if err != nil {
 			util.Exit(1, "Failed to load configuration: %s\n", err)
 		}

cmd/start.go

@@ -99,10 +99,9 @@ func runStartCmd(cmd *cobra.Command, args []string) {
 		s.smcache.RefreshLoop(ctx, time.Hour)
 	}()
 
-	// start workers
-	s.startCollector(ctx)
-	s.startVerifier(ctx)
-	s.startSubmitter(ctx)
+	// s.startCollector(ctx) -> Necessary event logs are directly fetched from the chain during verification
+	// s.startVerifier(ctx)-> Verificatin starts whenever the verse is discovered
+	// s.startSubmitter(ctx) -> Submission starts whenever the verse is discovered
 	s.startVerseDiscovery(ctx)
 	s.startBeacon(ctx)
 	log.Info("All workers started")
@@ -173,7 +172,7 @@ func mustNewServer(ctx context.Context) *server {
 		signers: map[string]ethutil.Signer{},
 	}
 
-	if s.conf, err = globalConfigLoader.load(); err != nil {
+	if s.conf, err = globalConfigLoader.load(true); err != nil {
 		log.Crit("Failed to load configuration", "err", err)
 	}
 
@@ -219,7 +218,10 @@ func (s *server) mustStartMetrics(ctx context.Context) {
 	go func() {
 		// NOTE: Don't add wait group, as no need to guarantee the completion
 		if err := metrics.ListenAndServe(ctx, s.msvr); err != nil {
-			log.Crit("Failed to start metrics server", "err", err)
+			// `ErrServerClosed` is thrown when `Shutdown` is intentionally called
+			if !errors.Is(err, http.ErrServerClosed) {
+				log.Crit("Failed to start metrics server", "err", err)
+			}
 		}
 		log.Info("Metrics server have exited listening", "addr", s.conf.Metrics.Listen)
 	}()
@@ -236,7 +238,10 @@ func (s *server) mustStartPprof(ctx context.Context) {
 	go func() {
 		// NOTE: Don't add wait group, as no need to guarantee the completion
 		if err := ps.ListenAndServe(ctx, s.psvr); err != nil {
-			log.Crit("Failed to start pprof server", "err", err)
+			// `ErrServerClosed` is thrown when `Shutdown` is intentionally called
+			if !errors.Is(err, http.ErrServerClosed) {
+				log.Crit("Failed to start pprof server", "err", err)
+			}
 		}
 		log.Info("pprof server have exited listening", "addr", s.conf.Debug.Pprof.Listen)
 	}()
@@ -297,8 +302,8 @@ func (s *server) mustStartP2P(ctx context.Context, ipc *ipc.IPCServer) {
 	go func() {
 		defer s.wg.Done()
 
-		s.p2p.Start(ctx)
-		log.Info("P2P node has stopped, decrement wait group")
+		enableSubscriber := s.conf.Submitter.Enable
+		s.p2p.Start(ctx, enableSubscriber)
 	}()
 }
 
@@ -358,7 +363,7 @@ func (s *server) mustSetupVerifier() {
 	}
 
 	l1Signer := ethutil.NewSignableClient(new(big.Int).SetUint64(s.conf.HubLayer.ChainID), s.hub, signer)
-	s.verifier = verifier.NewVerifier(&s.conf.Verifier, s.db, l1Signer)
+	s.verifier = verifier.NewVerifier(&s.conf.Verifier, s.db, s.p2p, l1Signer)
 }
 
 func (s *server) startVerifier(ctx context.Context) {
@@ -444,42 +449,56 @@ func (s *server) startSubmitter(ctx context.Context) {
 }
 
 func (s *server) startVerseDiscovery(ctx context.Context) {
+	if len(s.conf.VerseLayer.Directs) != 0 {
+		// read verses from the configuration
+		s.verseDiscoveryHandler(ctx, s.conf.VerseLayer.Directs)
+	}
+
 	if s.conf.VerseLayer.Discovery.Endpoint == "" {
-		// read verses from the configuration only, if the discovery endpoint is not set
-		s.verseDiscoveryHandler(s.conf.VerseLayer.Directs)
+		// Disable dinamically discovered verses, if the endpoint is not set
 		return
 	}
 
 	// dinamically discovered verses
-	disc := config.NewVerseDiscovery(
+	disc, err := config.NewVerseDiscovery(
+		ctx,
 		http.DefaultClient,
 		s.conf.VerseLayer.Discovery.Endpoint,
 		s.conf.VerseLayer.Discovery.RefreshInterval,
 	)
+	if err != nil {
+		log.Crit("Failed to construct verse discovery", "err", err)
+	}
+
+	// Subscribed verses to verifier and submitter
+	discSub := disc.Subscribe(ctx)
+
+	// synchronously try the first discovery
+	if err := disc.Work(ctx); err != nil {
+		// exit if the first discovery faild, because the following discovery highly likely fail
+		log.Crit("Failed to work verse discovery", "err", err)
+	}
 
 	s.wg.Add(1)
 	go func() {
 		defer func() {
 			defer s.wg.Done()
+			discSub.Cancel()
 			log.Info("Verse discovery has stopped, decrement wait group")
 		}()
 
 		discTick := time.NewTicker(s.conf.VerseLayer.Discovery.RefreshInterval)
 		defer discTick.Stop()
 
-		// Subscribed verses to verifier and submitter
-		sub := disc.Subscribe(ctx)
-		defer sub.Cancel()
-
 		log.Info("Verse discovery started", "endpoint", s.conf.VerseLayer.Discovery.Endpoint, "interval", s.conf.VerseLayer.Discovery.RefreshInterval)
 
 		for {
 			select {
 			case <-ctx.Done():
 				log.Info("Verse discovery stopped")
 				return
-			case verses := <-sub.Next():
-				s.verseDiscoveryHandler(verses)
+			case verses := <-discSub.Next():
+				s.verseDiscoveryHandler(ctx, verses)
 			case <-discTick.C:
 				if err := disc.Work(ctx); err != nil {
 					log.Error("Failed to work verse discovery", "err", err)
@@ -489,7 +508,7 @@ func (s *server) startVerseDiscovery(ctx context.Context) {
 	}()
 }
 
-func (s *server) verseDiscoveryHandler(discovers []*config.Verse) {
+func (s *server) verseDiscoveryHandler(ctx context.Context, discovers []*config.Verse) {
 	if s.verifier == nil && s.submitter == nil {
 		log.Warn("Both Verifier and Submitter are disabled")
 		return
@@ -509,7 +528,10 @@ func (s *server) verseDiscoveryHandler(discovers []*config.Verse) {
 		verse  verse.Verse
 		verify common.Address
 	}
-	var verses []*verse_
+	var (
+		verses        []*verse_
+		verseChainIDs []uint64
+	)
 	for _, cfg := range discovers {
 		for name, addr := range cfg.L1Contracts {
 			if factory, ok := verseFactories[name]; ok {
@@ -518,18 +540,23 @@ func (s *server) verseDiscoveryHandler(discovers []*config.Verse) {
 					verse:  factory(s.db, s.hub, common.HexToAddress(addr)),
 					verify: verifyContracts[name],
 				})
+				verseChainIDs = append(verseChainIDs, cfg.ChainID)
 			}
 		}
 	}
 
+	log.Info("Discovered verses", "count", len(verses), "chain-ids", verseChainIDs)
+
 	for _, x := range verses {
 		// add verse to Verifier
 		if s.verifier != nil && !s.verifier.HasTask(x.verse.RollupContract(), x.cfg.RPC) {
 			l2Client, err := ethutil.NewClient(x.cfg.RPC)
 			if err != nil {
 				log.Error("Failed to construct verse-layer client", "err", err)
 			} else {
-				s.verifier.AddTask(x.verse.WithVerifiable(l2Client))
+				log.Info("Add verse to Verifier", "chain-id", x.cfg.ChainID, "contract", x.verse.RollupContract())
+				// s.verifier.AddTask(x.verse.WithVerifiable(l2Client))
+				s.verifier.AddVerse(ctx, x.verse.WithVerifiable(l2Client), x.cfg.ChainID)
 			}
 		}
 
@@ -546,8 +573,10 @@ func (s *server) verseDiscoveryHandler(discovers []*config.Verse) {
 					continue
 				}
 
+				log.Info("Add verse to Submitter", "chain-id", x.cfg.ChainID, "contract", x.verse.RollupContract())
 				l1Signer := ethutil.NewSignableClient(new(big.Int).SetUint64(s.conf.HubLayer.ChainID), s.hub, signer)
-				s.submitter.AddTask(x.verse.WithTransactable(l1Signer, x.verify))
+				// s.submitter.AddTask(x.verse.WithTransactable(l1Signer, x.verify))
+				s.submitter.AddVerse(ctx, x.verse.WithTransactable(l1Signer, x.verify), x.cfg.ChainID)
 			}
 		}
 	}

cmd/status.go

@@ -11,7 +11,7 @@ var statusCmd = &cobra.Command{
 	Short: "Show status",
 	Long:  "Show status",
 	Run: func(cmd *cobra.Command, args []string) {
-		conf, err := globalConfigLoader.load()
+		conf, err := globalConfigLoader.load(true)
 		if err != nil {
 			util.Exit(1, "Failed to load configuration: %s\n", err)
 		}

cmd/unlock.go

@@ -32,7 +32,7 @@ func init() {
 }
 
 func runUnlockCmd(cmd *cobra.Command, args []string) {
-	conf, err := globalConfigLoader.load()
+	conf, err := globalConfigLoader.load(true)
 	if err != nil {
 		util.Exit(1, "Failed to load configuration: %s\n", err)
 	}

collector/event_collector.go

@@ -71,7 +71,7 @@ func (w *EventCollector) Work(ctx context.Context) {
 
 		// collect event logs from hub-layer
 		start, end := blocks[0], blocks[len(blocks)-1]
-		logs, err := w.hub.FilterLogs(ctx, verse.NewEventLogFilter(start.Number, end.Number))
+		logs, err := w.hub.FilterLogs(ctx, verse.NewEventLogFilter(start.Number, end.Number, nil))
 		if err != nil {
 			w.log.Error("Failed to fetch event logs from hub-layer",
 				"start", start, "end", end, "err", err)