Rework exitwhensynced functionality

cmd/opera/launcher/config.go

@@ -98,6 +98,14 @@ var (
 		Usage: `Blockchain sync mode ("full" or "snap")`,
 		Value: "full",
 	}
+	ExitWhenAgeFlag = cli.DurationFlag{
+		Name:  "exitwhensynced.age",
+		Usage: "Exits after synchronisation reaches the required age",
+	}
+	ExitWhenEpochFlag = cli.Uint64Flag{
+		Name:  "exitwhensynced.epoch",
+		Usage: "Exits after synchronisation reaches the required epoch",
+	}
 )
 
 type GenesisTemplate struct {

cmd/opera/launcher/launcher.go

@@ -1,14 +1,14 @@
 package launcher
 
 import (
-	"context"
 	"fmt"
 	"os"
 	"path"
 	"sort"
 	"strings"
 	"time"
 
+	"github.com/Fantom-foundation/lachesis-base/inter/idx"
 	"github.com/ethereum/go-ethereum/accounts"
 	"github.com/ethereum/go-ethereum/accounts/keystore"
 	"github.com/ethereum/go-ethereum/cmd/utils"
@@ -112,7 +112,8 @@ func initFlags() {
 		utils.KeyStoreDirFlag,
 		utils.USBFlag,
 		utils.SmartCardDaemonPathFlag,
-		utils.ExitWhenSyncedFlag,
+		ExitWhenAgeFlag,
+		ExitWhenEpochFlag,
 		utils.LightKDFFlag,
 		configFileFlag,
 		validatorIDFlag,
@@ -321,7 +322,19 @@ func makeNode(ctx *cli.Context, cfg *config, genesisStore *genesisstore.Store) (
 		}
 		return evmcore.NewTxPool(cfg.TxPool, reader.Config(), reader)
 	}
-	svc, err := gossip.NewService(stack, cfg.Opera, gdb, blockProc, engine, dagIndex, newTxPool)
+	haltCheck := func(oldEpoch, newEpoch idx.Epoch, age time.Time) bool {
+		stop := ctx.GlobalIsSet(ExitWhenAgeFlag.Name) && ctx.GlobalDuration(ExitWhenAgeFlag.Name) >= time.Since(age)
+		stop = stop || ctx.GlobalIsSet(ExitWhenEpochFlag.Name) && idx.Epoch(ctx.GlobalUint64(ExitWhenEpochFlag.Name)) <= newEpoch
+		if stop {
+			go func() {
+				// do it in a separate thread to avoid deadlock
+				_ = stack.Close()
+			}()
+			return true
+		}
+		return false
+	}
+	svc, err := gossip.NewService(stack, cfg.Opera, gdb, blockProc, engine, dagIndex, newTxPool, haltCheck)
 	if err != nil {
 		utils.Fatalf("Failed to create the service: %v", err)
 	}
@@ -406,34 +419,6 @@ func startNode(ctx *cli.Context, stack *node.Node) {
 			}
 		}
 	}()
-
-	// Spawn a standalone goroutine for status synchronization monitoring,
-	// close the node when synchronization is complete if user required.
-	if ctx.GlobalBool(utils.ExitWhenSyncedFlag.Name) {
-		go func() {
-			for first := true; ; first = false {
-				// Call ftm_syncing until it returns false
-				time.Sleep(5 * time.Second)
-
-				var syncing bool
-				err := rpcClient.CallContext(context.TODO(), &syncing, "ftm_syncing")
-				if err != nil {
-					continue
-				}
-				if !syncing {
-					if !first {
-						time.Sleep(time.Minute)
-					}
-					log.Info("Synchronisation completed. Exiting due to exitwhensynced flag.")
-					err = stack.Close()
-					if err != nil {
-						continue
-					}
-					return
-				}
-			}
-		}()
-	}
 }
 
 // unlockAccounts unlocks any account specifically requested.

gossip/c_event_callbacks.go

@@ -237,6 +237,11 @@ func (s *Service) processEvent(e *inter.EventPayload) error {
 	}
 
 	s.mayCommit(newEpoch != oldEpoch)
+
+	if s.haltCheck != nil && s.haltCheck(oldEpoch, newEpoch, e.MedianTime().Time()) {
+		// halt syncing
+		s.stopped = true
+	}
 	return nil
 }
 

gossip/service.go

@@ -149,14 +149,17 @@ type Service struct {
 
 	procLogger *proclogger.Logger
 
-	stopped bool
+	stopped      bool
+	haltCheck func(oldEpoch, newEpoch idx.Epoch, time time.Time) bool
 
 	tflusher PeriodicFlusher
 
 	logger.Instance
 }
 
-func NewService(stack *node.Node, config Config, store *Store, blockProc BlockProc, engine lachesis.Consensus, dagIndexer *vecmt.Index, newTxPool func(evmcore.StateReader) TxPool) (*Service, error) {
+func NewService(stack *node.Node, config Config, store *Store, blockProc BlockProc,
+	engine lachesis.Consensus, dagIndexer *vecmt.Index, newTxPool func(evmcore.StateReader) TxPool,
+	haltCheck func(oldEpoch, newEpoch idx.Epoch, age time.Time) bool) (*Service, error) {
 	if err := config.Validate(); err != nil {
 		return nil, err
 	}
@@ -171,6 +174,7 @@ func NewService(stack *node.Node, config Config, store *Store, blockProc BlockPr
 	svc.eventMux = stack.EventMux()
 	// Create the net API service
 	svc.netRPCService = ethapi.NewPublicNetAPI(svc.p2pServer, store.GetRules().NetworkID)
+	svc.haltCheck = haltCheck
 
 	return svc, nil
 }
@@ -444,6 +448,11 @@ func (s *Service) Start() error {
 
 	s.verWatcher.Start()
 
+	if s.haltCheck != nil && s.haltCheck(s.store.GetEpoch(), s.store.GetEpoch(), s.store.GetBlockState().LastBlock.Time.Time()) {
+		// halt syncing
+		s.stopped = true
+	}
+
 	return nil
 }