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

added sync monitor for api keyper

Author: blockchainluffy

.circleci/config.yml

@@ -156,7 +156,7 @@ jobs:
           name: "Run unit tests with gotestsum"
           command: |
             mkdir -p report/unit report/integration
-            gotestsum -f standard-verbose --junitfile report/unit/tests.xml -- -short -race ./...
+            gotestsum -f standard-verbose --junitfile report/unit/tests.xml -- -short -race -count=1 -p 1 ./...
       - run:
           name: Wait for PostgreSQL
           command: |

rolling-shutter/keyperimpl/shutterservice/config.go

@@ -81,16 +81,18 @@ func (c *Config) GetAddress() common.Address {
 }
 
 type ChainConfig struct {
-	Node                 *configuration.EthnodeConfig `shconfig:",required"`
-	Contracts            *ContractsConfig             `shconfig:",required"`
-	SyncStartBlockNumber uint64                       `shconfig:",required"`
+	Node                     *configuration.EthnodeConfig `shconfig:",required"`
+	Contracts                *ContractsConfig             `shconfig:",required"`
+	SyncStartBlockNumber     uint64                       `shconfig:",required"`
+	SyncMonitorCheckInterval uint64                       `shconfig:",required"`
 }
 
 func NewChainConfig() *ChainConfig {
 	c := &ChainConfig{
-		Node:                 configuration.NewEthnodeConfig(),
-		Contracts:            NewContractsConfig(),
-		SyncStartBlockNumber: 0,
+		Node:                     configuration.NewEthnodeConfig(),
+		Contracts:                NewContractsConfig(),
+		SyncStartBlockNumber:     0,
+		SyncMonitorCheckInterval: 0,
 	}
 	c.Init()
 	return c
@@ -111,10 +113,12 @@ func (c *ChainConfig) Validate() error {
 
 func (c *ChainConfig) SetDefaultValues() error {
 	c.SyncStartBlockNumber = 0
+	c.SyncMonitorCheckInterval = 30
 	return c.Contracts.SetDefaultValues()
 }
 
 func (c *ChainConfig) SetExampleValues() error {
+	c.SyncMonitorCheckInterval = 30
 	return nil
 }
 

rolling-shutter/keyperimpl/shutterservice/keyper.go

@@ -2,6 +2,7 @@ package shutterservice
 
 import (
 	"context"
+	"time"
 
 	"github.com/ethereum/go-ethereum/ethclient"
 	gethLog "github.com/ethereum/go-ethereum/log"
@@ -35,6 +36,7 @@ type Keyper struct {
 	registrySyncer      *RegistrySyncer
 	eonKeyPublisher     *eonkeypublisher.EonKeyPublisher
 	latestTriggeredTime *uint64
+	syncMonitor         *SyncMonitor
 
 	// input events
 	newBlocks        chan *syncevent.LatestBlock
@@ -111,8 +113,12 @@ func (kpr *Keyper) Start(ctx context.Context, runner service.Runner) error {
 		return err
 	}
 
+	kpr.syncMonitor = &SyncMonitor{
+		DBPool:        kpr.dbpool,
+		CheckInterval: time.Duration(kpr.config.Chain.SyncMonitorCheckInterval) * time.Second,
+	}
 	runner.Go(func() error { return kpr.processInputs(ctx) })
-	return runner.StartService(kpr.core, kpr.chainSyncClient, kpr.eonKeyPublisher)
+	return runner.StartService(kpr.core, kpr.chainSyncClient, kpr.eonKeyPublisher, kpr.syncMonitor)
 }
 
 func NewKeyper(kpr *Keyper, messagingMiddleware *MessagingMiddleware) (*keyper.KeyperCore, error) {

rolling-shutter/keyperimpl/shutterservice/syncmonitor.go

@@ -0,0 +1,64 @@
+package shutterservice
+
+import (
+	"context"
+	"time"
+
+	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog/log"
+
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/shutterservice/database"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
+)
+
+type SyncMonitor struct {
+	DBPool        *pgxpool.Pool
+	CheckInterval time.Duration
+}
+
+func (s *SyncMonitor) Start(ctx context.Context, runner service.Runner) error {
+	runner.Go(func() error {
+		return s.runMonitor(ctx)
+	})
+
+	return nil
+}
+
+func (s *SyncMonitor) runMonitor(ctx context.Context) error {
+	var lastBlockNumber int64
+	db := database.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
+				}
+				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")
+			}
+		case <-ctx.Done():
+			log.Info().Msg("stopping syncMonitor due to context cancellation")
+			return ctx.Err()
+		}
+	}
+}

rolling-shutter/keyperimpl/shutterservice/syncmonitor_test.go

@@ -0,0 +1,143 @@
+package shutterservice_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"gotest.tools/assert"
+
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/shutterservice"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/shutterservice/database"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/testsetup"
+)
+
+func TestAPISyncMonitor_ThrowsErrorWhenBlockNotIncreasing(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	dbpool, dbclose := testsetup.NewTestDBPool(ctx, t, database.Definition)
+	defer dbclose()
+	db := database.New(dbpool)
+
+	initialBlockNumber := int64(100)
+
+	err := db.SetIdentityRegisteredEventSyncedUntil(ctx, database.SetIdentityRegisteredEventSyncedUntilParams{
+		BlockHash:   []byte{0x01, 0x02, 0x03},
+		BlockNumber: initialBlockNumber,
+	})
+	if err != nil {
+		t.Fatalf("failed to set initial synced data: %v", err)
+	}
+
+	monitor := &shutterservice.SyncMonitor{
+		DBPool:        dbpool,
+		CheckInterval: 5 * time.Second,
+	}
+
+	errCh := make(chan error, 1)
+
+	go func() {
+		err := service.RunWithSighandler(ctx, monitor)
+		if err != nil {
+			errCh <- err
+		}
+	}()
+
+	time.Sleep(12 * time.Second)
+
+	select {
+	case err := <-errCh:
+		assert.ErrorContains(t, err, "block number has not increased between checks")
+	case <-time.After(5 * time.Second):
+		t.Fatal("expected an error, but none was returned")
+	}
+}
+
+func TestAPISyncMonitor_HandlesBlockNumberIncreasing(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	dbpool, closeDB := testsetup.NewTestDBPool(ctx, t, database.Definition)
+	defer closeDB()
+	db := database.New(dbpool)
+
+	initialBlockNumber := int64(100)
+	err := db.SetIdentityRegisteredEventSyncedUntil(ctx, database.SetIdentityRegisteredEventSyncedUntilParams{
+		BlockHash:   []byte{0x01, 0x02, 0x03},
+		BlockNumber: initialBlockNumber,
+	})
+	if err != nil {
+		t.Fatalf("failed to set initial synced data: %v", err)
+	}
+
+	monitor := &shutterservice.SyncMonitor{
+		DBPool:        dbpool,
+		CheckInterval: 5 * time.Second,
+	}
+
+	_, deferFn := service.RunBackground(ctx, monitor)
+	defer deferFn()
+
+	doneCh := make(chan struct{})
+	go func() {
+		for i := 0; i < 5; i++ {
+			newBlockNumber := initialBlockNumber + int64(i+1)
+			err := db.SetIdentityRegisteredEventSyncedUntil(ctx, database.SetIdentityRegisteredEventSyncedUntilParams{
+				BlockHash:   []byte{0x01, 0x02, 0x03},
+				BlockNumber: newBlockNumber,
+			})
+			if err != nil {
+				t.Errorf("failed to update block number: %v", err)
+				return
+			}
+
+			time.Sleep(5 * time.Second)
+		}
+
+		doneCh <- struct{}{}
+	}()
+
+	<-doneCh
+	syncedData, err := db.GetIdentityRegisteredEventsSyncedUntil(ctx)
+	if err != nil {
+		t.Fatalf("failed to retrieve final block number: %v", err)
+	}
+
+	assert.Equal(t, initialBlockNumber+5, syncedData.BlockNumber, "block number should have been incremented correctly")
+}
+
+func TestAPISyncMonitor_ContinuesWhenNoRows(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	dbpool, closeDB := testsetup.NewTestDBPool(ctx, t, database.Definition)
+	defer closeDB()
+	_ = database.New(dbpool)
+
+	monitor := &shutterservice.SyncMonitor{
+		DBPool:        dbpool,
+		CheckInterval: 5 * time.Second,
+	}
+
+	monitorCtx, cancelMonitor := context.WithCancel(ctx)
+	defer cancelMonitor()
+
+	errCh := make(chan error, 1)
+	go func() {
+		err := service.RunWithSighandler(monitorCtx, monitor)
+		if err != nil {
+			errCh <- err
+		}
+	}()
+
+	time.Sleep(15 * time.Second)
+	cancelMonitor()
+
+	select {
+	case err := <-errCh:
+		t.Fatalf("expected monitor to continue without error, but got: %v", err)
+	case <-time.After(1 * time.Second):
+	}
+}