Merge pull request #584 from shutter-network/feat/api-sync-monitor

updated sync monitor to not throw error when dkg is happening

Author: blockchainluffy

rolling-shutter/keyperimpl/shutterservice/metrics.go

@@ -0,0 +1,16 @@
+package shutterservice
+
+import "github.com/prometheus/client_golang/prometheus"
+
+var metricsRegistryEventsSyncedUntil = prometheus.NewGauge(
+	prometheus.GaugeOpts{
+		Namespace: "shutter",
+		Subsystem: "shutter_api",
+		Name:      "registry_events_syned_until",
+		Help:      "Current value of the latest block fetched",
+	},
+)
+
+func init() {
+	prometheus.MustRegister(metricsRegistryEventsSyncedUntil)
+}

rolling-shutter/keyperimpl/shutterservice/registrysyncer.go

@@ -182,6 +182,8 @@ func (s *RegistrySyncer) syncRange(
 		Int("num-inserted-events", len(filteredEvents)).
 		Int("num-discarded-events", len(events)-len(filteredEvents)).
 		Msg("synced registry contract")
+
+	metricsRegistryEventsSyncedUntil.Set(float64(end))
 	return nil
 }
 

rolling-shutter/keyperimpl/shutterservice/syncmonitor.go

@@ -2,13 +2,15 @@ package shutterservice
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/jackc/pgx/v4"
 	"github.com/jackc/pgx/v4/pgxpool"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
 
+	keyperDB "github.com/shutter-network/rolling-shutter/rolling-shutter/keyper/database"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/shutterservice/database"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
 )
@@ -29,36 +31,91 @@ func (s *SyncMonitor) Start(ctx context.Context, runner service.Runner) error {
 func (s *SyncMonitor) runMonitor(ctx context.Context) error {
 	var lastBlockNumber int64
 	db := database.New(s.DBPool)
+	keyperdb := keyperDB.New(s.DBPool)
 
 	log.Debug().Msg("starting the sync monitor")
 
 	for {
 		select {
 		case <-time.After(s.CheckInterval):
-			record, err := db.GetIdentityRegisteredEventsSyncedUntil(ctx)
-			if err != nil {
-				if errors.Is(err, pgx.ErrNoRows) {
-					log.Warn().Err(err).Msg("no rows found in table identity_registered_events_synced_until")
-					continue
+			if err := s.runCheck(ctx, db, keyperdb, &lastBlockNumber); err != nil {
+				if errors.Is(err, ErrBlockNotIncreasing) {
+					return err
 				}
-				return errors.Wrap(err, "error getting identity_registered_events_synced_until")
-			}
-
-			currentBlockNumber := record.BlockNumber
-			log.Debug().Int64("current-block-number", currentBlockNumber).Msg("current block number")
-
-			if currentBlockNumber > lastBlockNumber {
-				lastBlockNumber = currentBlockNumber
-			} else {
-				log.Error().
-					Int64("last-block-number", lastBlockNumber).
-					Int64("current-block-number", currentBlockNumber).
-					Msg("block number has not increased between checks")
-				return errors.New("block number has not increased between checks")
+				log.Debug().Err(err).Msg("skipping sync check due to error")
 			}
 		case <-ctx.Done():
 			log.Info().Msg("stopping syncMonitor due to context cancellation")
 			return ctx.Err()
 		}
 	}
 }
+
+var ErrBlockNotIncreasing = errors.New("block number has not increased between checks")
+
+func (s *SyncMonitor) runCheck(
+	ctx context.Context,
+	db *database.Queries,
+	keyperdb *keyperDB.Queries,
+	lastBlockNumber *int64,
+) error {
+	isRunning, err := s.isDKGRunning(ctx, keyperdb)
+	if err != nil {
+		return fmt.Errorf("syncMonitor | error in dkgIsRunning: %w", err)
+	}
+	if isRunning {
+		log.Debug().Msg("dkg is running, skipping sync monitor checks")
+		return nil
+	}
+
+	record, err := db.GetIdentityRegisteredEventsSyncedUntil(ctx)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			log.Warn().Err(err).Msg("no rows found in table identity_registered_events_synced_until")
+			return nil // This is not an error condition that should stop monitoring
+		}
+		return fmt.Errorf("error getting identity_registered_events_synced_until: %w", err)
+	}
+
+	currentBlockNumber := record.BlockNumber
+	log.Debug().Int64("current-block-number", currentBlockNumber).Msg("current block number")
+
+	if currentBlockNumber > *lastBlockNumber {
+		*lastBlockNumber = currentBlockNumber
+		return nil
+	}
+
+	log.Error().
+		Int64("last-block-number", *lastBlockNumber).
+		Int64("current-block-number", currentBlockNumber).
+		Msg("block number has not increased between checks")
+	return ErrBlockNotIncreasing
+}
+
+func (s *SyncMonitor) isDKGRunning(ctx context.Context, keyperdb *keyperDB.Queries) (bool, error) {
+	// if latest eon is registered then EonStarted event has triggered, which means the dkg can start
+	eons, err := keyperdb.GetAllEons(ctx)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		log.Error().
+			Err(err).
+			Msg("syncMonitor | error getting all eons")
+		return false, err
+	}
+
+	if len(eons) == 0 {
+		return false, nil
+	}
+
+	// if we get no rows in getting dkg result then dkg is not completed for that eon
+	_, err = keyperdb.GetDKGResult(ctx, eons[len(eons)-1].Eon)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return true, nil
+	} else if err != nil {
+		log.Error().Err(err).Msg("syncMonitor | error getting dkg result")
+		return false, err
+	}
+	return false, nil
+}