Merge pull request #559 from shutter-network/feat/monitor-event-synch

Add last event synched monitor

Author: ylembachar

.gitignore

@@ -8,4 +8,7 @@ log
 node_modules
 result
 temp
-work
+work
+.env
+.idea
+.DS_Store

play/manual/keyper-0.toml

@@ -15,6 +15,9 @@ CustomBootstrapAddresses = ['/ip4/127.0.0.1/tcp/2001/p2p/12D3KooWMyutShWdqYj7fre
 Environment = "local"
 DiscoveryNamespace = "shutter-42"
 
+[P2P.FloodSubDiscovery]
+enabled = false
+
 [Gnosis]
 EncryptedGasLimit = 1000000
 MinGasPerTransaction = 21000
@@ -23,6 +26,7 @@ SecondsPerSlot = 5
 SlotsPerEpoch = 16
 GenesisSlotTimestamp = 1665396300
 SyncStartBlockNumber = 0
+SyncMonitorCheckInterval = 30
 
 [Gnosis.Node]
 PrivateKey = 'dcb23da56656b3c8a11e2b4cdd92f430c500862f7f7fc762807d11b734e9500c'

play/manual/keyper-1.toml

@@ -15,6 +15,9 @@ CustomBootstrapAddresses = ['/ip4/127.0.0.1/tcp/2001/p2p/12D3KooWMyutShWdqYj7fre
 Environment = "local"
 DiscoveryNamespace = "shutter-42"
 
+[P2P.FloodSubDiscovery]
+enabled = false
+
 [Gnosis]
 EncryptedGasLimit = 1000000
 MinGasPerTransaction = 21000
@@ -23,6 +26,7 @@ SecondsPerSlot = 5
 SlotsPerEpoch = 16
 GenesisSlotTimestamp = 1665396300
 SyncStartBlockNumber = 0
+SyncMonitorCheckInterval = 30
 
 [Gnosis.Node]
 PrivateKey = '00b4a53228e3761ad78bd376b8293f19af36777f71d8e55a61975f8eecd8c1c1'

play/manual/keyper-2.toml

@@ -15,6 +15,9 @@ CustomBootstrapAddresses = ['/ip4/127.0.0.1/tcp/2001/p2p/12D3KooWMyutShWdqYj7fre
 Environment = "local"
 DiscoveryNamespace = "shutter-42"
 
+[P2P.FloodSubDiscovery]
+enabled = false
+
 [Gnosis]
 EncryptedGasLimit = 1000000
 MinGasPerTransaction = 21000
@@ -23,6 +26,7 @@ SecondsPerSlot = 5
 SlotsPerEpoch = 16
 GenesisSlotTimestamp = 1665396300
 SyncStartBlockNumber = 0
+SyncMonitorCheckInterval = 30
 
 [Gnosis.Node]
 PrivateKey = 'a4e901b1df81ff8fc5fa77f5bf0c15a4c8410e85fcaf19fbec47a2241b9d65d6'

rolling-shutter/keyperimpl/gnosis/config.go

@@ -98,28 +98,30 @@ func (c *Config) GetAddress() common.Address {
 }
 
 type GnosisConfig struct {
-	Node                 *configuration.EthnodeConfig `shconfig:",required"`
-	Contracts            *GnosisContractsConfig       `shconfig:",required"`
-	EncryptedGasLimit    uint64                       `shconfig:",required"`
-	MinGasPerTransaction uint64                       `shconfig:",required"`
-	MaxTxPointerAge      uint64                       `shconfig:",required"`
-	SecondsPerSlot       uint64                       `shconfig:",required"`
-	SlotsPerEpoch        uint64                       `shconfig:",required"`
-	GenesisSlotTimestamp uint64                       `shconfig:",required"`
-	SyncStartBlockNumber uint64                       `shconfig:",required"`
+	Node                     *configuration.EthnodeConfig `shconfig:",required"`
+	Contracts                *GnosisContractsConfig       `shconfig:",required"`
+	EncryptedGasLimit        uint64                       `shconfig:",required"`
+	MinGasPerTransaction     uint64                       `shconfig:",required"`
+	MaxTxPointerAge          uint64                       `shconfig:",required"`
+	SecondsPerSlot           uint64                       `shconfig:",required"`
+	SlotsPerEpoch            uint64                       `shconfig:",required"`
+	GenesisSlotTimestamp     uint64                       `shconfig:",required"`
+	SyncStartBlockNumber     uint64                       `shconfig:",required"`
+	SyncMonitorCheckInterval uint64                       `shconfig:",required"`
 }
 
 func NewGnosisConfig() *GnosisConfig {
 	c := &GnosisConfig{
-		Node:                 configuration.NewEthnodeConfig(),
-		Contracts:            NewGnosisContractsConfig(),
-		EncryptedGasLimit:    0,
-		MinGasPerTransaction: 0,
-		MaxTxPointerAge:      0,
-		SecondsPerSlot:       0,
-		SlotsPerEpoch:        0,
-		GenesisSlotTimestamp: 0,
-		SyncStartBlockNumber: 0,
+		Node:                     configuration.NewEthnodeConfig(),
+		Contracts:                NewGnosisContractsConfig(),
+		EncryptedGasLimit:        0,
+		MinGasPerTransaction:     0,
+		MaxTxPointerAge:          0,
+		SecondsPerSlot:           0,
+		SlotsPerEpoch:            0,
+		GenesisSlotTimestamp:     0,
+		SyncStartBlockNumber:     0,
+		SyncMonitorCheckInterval: 0,
 	}
 	c.Init()
 	return c
@@ -156,6 +158,7 @@ func (c *GnosisConfig) SetExampleValues() error {
 	c.SlotsPerEpoch = 16
 	c.GenesisSlotTimestamp = 1665410700
 	c.SyncStartBlockNumber = 0
+	c.SyncMonitorCheckInterval = 30
 	return nil
 }
 

rolling-shutter/keyperimpl/gnosis/keyper.go

@@ -49,6 +49,7 @@ type Keyper struct {
 	validatorSyncer     *ValidatorSyncer
 	eonKeyPublisher     *eonkeypublisher.EonKeyPublisher
 	latestTriggeredSlot *uint64
+	syncMonitor         *SyncMonitor
 
 	// input events
 	newBlocks        chan *syncevent.LatestBlock
@@ -62,7 +63,8 @@ type Keyper struct {
 
 func New(c *Config) *Keyper {
 	return &Keyper{
-		config: c,
+		config:      c,
+		syncMonitor: &SyncMonitor{},
 	}
 }
 
@@ -154,8 +156,13 @@ func (kpr *Keyper) Start(ctx context.Context, runner service.Runner) error {
 		return errors.Wrap(err, "failed to reset transaction pointer age")
 	}
 
+	kpr.syncMonitor = &SyncMonitor{
+		DBPool:        kpr.dbpool,
+		CheckInterval: time.Duration(kpr.config.Gnosis.SyncMonitorCheckInterval) * time.Second,
+	}
+
 	runner.Go(func() error { return kpr.processInputs(ctx) })
-	return runner.StartService(kpr.core, kpr.chainSyncClient, kpr.slotTicker, kpr.eonKeyPublisher)
+	return runner.StartService(kpr.core, kpr.chainSyncClient, kpr.slotTicker, kpr.eonKeyPublisher, kpr.syncMonitor)
 }
 
 func NewKeyper(kpr *Keyper, messagingMiddleware *MessagingMiddleware) (*keyper.KeyperCore, error) {

rolling-shutter/keyperimpl/gnosis/syncmonitor.go

@@ -0,0 +1,64 @@
+package gnosis
+
+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/gnosis/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.GetTransactionSubmittedEventsSyncedUntil(ctx)
+			if err != nil {
+				if errors.Is(err, pgx.ErrNoRows) {
+					log.Warn().Err(err).Msg("no rows found in table transaction_submitted_events_synced_until")
+					continue
+				}
+				return errors.Wrap(err, "error getting transaction_submitted_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/gnosis/syncmonitor_test.go

@@ -0,0 +1,146 @@
+package gnosis_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"gotest.tools/assert"
+
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/gnosis"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/gnosis/database"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/testsetup"
+)
+
+func TestSyncMonitor_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.SetTransactionSubmittedEventsSyncedUntil(ctx, database.SetTransactionSubmittedEventsSyncedUntilParams{
+		BlockHash:   []byte{0x01, 0x02, 0x03},
+		BlockNumber: initialBlockNumber,
+		Slot:        1,
+	})
+	if err != nil {
+		t.Fatalf("failed to set initial synced data: %v", err)
+	}
+
+	monitor := &gnosis.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 TestSyncMonitor_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.SetTransactionSubmittedEventsSyncedUntil(ctx, database.SetTransactionSubmittedEventsSyncedUntilParams{
+		BlockHash:   []byte{0x01, 0x02, 0x03},
+		BlockNumber: initialBlockNumber,
+		Slot:        1,
+	})
+	if err != nil {
+		t.Fatalf("failed to set initial synced data: %v", err)
+	}
+
+	monitor := &gnosis.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.SetTransactionSubmittedEventsSyncedUntil(ctx, database.SetTransactionSubmittedEventsSyncedUntilParams{
+				BlockHash:   []byte{0x01, 0x02, 0x03},
+				BlockNumber: newBlockNumber,
+				Slot:        1,
+			})
+			if err != nil {
+				t.Errorf("failed to update block number: %v", err)
+				return
+			}
+
+			time.Sleep(5 * time.Second)
+		}
+
+		doneCh <- struct{}{}
+	}()
+
+	<-doneCh
+	syncedData, err := db.GetTransactionSubmittedEventsSyncedUntil(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 TestSyncMonitor_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 := &gnosis.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):
+	}
+}