updated sync monitor to not throw error when dkg is happening

Author: blockchainluffy

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,68 @@ 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 {
+	batchConfig, err := keyperdb.GetLatestBatchConfig(ctx)
+	if err != nil {
+		return fmt.Errorf("error getting batchconfig: %w", err)
+	}
+
+	dkgResult, err := keyperdb.GetDKGResult(ctx, int64(batchConfig.KeyperConfigIndex))
+	if err != nil {
+		return fmt.Errorf("error getting dkgresult: %w", err)
+	}
+
+	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 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
+		return nil
+	}
+
+	if dkgResult.Success {
+		log.Error().
+			Int64("last-block-number", *lastBlockNumber).
+			Int64("current-block-number", currentBlockNumber).
+			Msg("block number has not increased between checks")
+		return ErrBlockNotIncreasing
+	}
+
+	return nil
+}

rolling-shutter/keyperimpl/shutterservice/syncmonitor_test.go

@@ -7,37 +7,56 @@ import (
 
 	"gotest.tools/assert"
 
+	"github.com/jackc/pgx/v4/pgxpool"
+	keyperDB "github.com/shutter-network/rolling-shutter/rolling-shutter/keyper/database"
 	"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 setupTestData(ctx context.Context, t *testing.T, dbpool *pgxpool.Pool, blockNumber int64) {
+	db := database.New(dbpool)
+	keyperdb := keyperDB.New(dbpool)
+
+	// Set up batch config
+	err := keyperdb.InsertBatchConfig(ctx, keyperDB.InsertBatchConfigParams{
+		KeyperConfigIndex: 1,
+		Keypers:           []string{},
+	})
+	assert.NilError(t, err)
+
+	// Set up DKG result
+	err = keyperdb.InsertDKGResult(ctx, keyperDB.InsertDKGResultParams{
+		Eon:     1,
+		Success: true,
+	})
+	assert.NilError(t, err)
+
+	// Set up initial block
+	err = db.SetIdentityRegisteredEventSyncedUntil(ctx, database.SetIdentityRegisteredEventSyncedUntilParams{
+		BlockHash:   []byte{0x01, 0x02, 0x03},
+		BlockNumber: blockNumber,
+	})
+	assert.NilError(t, err)
+}
+
 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)
-	}
+	setupTestData(ctx, t, dbpool, initialBlockNumber)
 
 	monitor := &shutterservice.SyncMonitor{
 		DBPool:        dbpool,
 		CheckInterval: 5 * time.Second,
 	}
 
 	errCh := make(chan error, 1)
-
 	go func() {
 		err := service.RunWithSighandler(ctx, monitor)
 		if err != nil {
@@ -49,7 +68,7 @@ func TestAPISyncMonitor_ThrowsErrorWhenBlockNotIncreasing(t *testing.T) {
 
 	select {
 	case err := <-errCh:
-		assert.ErrorContains(t, err, "block number has not increased between checks")
+		assert.ErrorContains(t, err, shutterservice.ErrBlockNotIncreasing.Error())
 	case <-time.After(5 * time.Second):
 		t.Fatal("expected an error, but none was returned")
 	}
@@ -64,13 +83,7 @@ func TestAPISyncMonitor_HandlesBlockNumberIncreasing(t *testing.T) {
 	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)
-	}
+	setupTestData(ctx, t, dbpool, initialBlockNumber)
 
 	monitor := &shutterservice.SyncMonitor{
 		DBPool:        dbpool,
@@ -114,7 +127,21 @@ func TestAPISyncMonitor_ContinuesWhenNoRows(t *testing.T) {
 
 	dbpool, closeDB := testsetup.NewTestDBPool(ctx, t, database.Definition)
 	defer closeDB()
-	_ = database.New(dbpool)
+
+	// Only set up keyper set and DKG result, but no block data
+	keyperdb := keyperDB.New(dbpool)
+
+	err := keyperdb.InsertBatchConfig(ctx, keyperDB.InsertBatchConfigParams{
+		KeyperConfigIndex: 1,
+		Keypers:           []string{},
+	})
+	assert.NilError(t, err)
+
+	err = keyperdb.InsertDKGResult(ctx, keyperDB.InsertDKGResultParams{
+		Eon:     1,
+		Success: true,
+	})
+	assert.NilError(t, err)
 
 	monitor := &shutterservice.SyncMonitor{
 		DBPool:        dbpool,
@@ -141,3 +168,109 @@ func TestAPISyncMonitor_ContinuesWhenNoRows(t *testing.T) {
 	case <-time.After(1 * time.Second):
 	}
 }
+
+func TestAPISyncMonitor_ContinuesWhenNoDKGResult(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)
+	keyperdb := keyperDB.New(dbpool)
+
+	// Set up batch config, but no DKG result
+	err := keyperdb.InsertBatchConfig(ctx, keyperDB.InsertBatchConfigParams{
+		KeyperConfigIndex: 1,
+		Keypers:           []string{},
+	})
+	assert.NilError(t, err)
+
+	// Set up initial block data
+	initialBlockNumber := int64(100)
+	err = db.SetIdentityRegisteredEventSyncedUntil(ctx, database.SetIdentityRegisteredEventSyncedUntilParams{
+		BlockHash:   []byte{0x01, 0x02, 0x03},
+		BlockNumber: initialBlockNumber,
+	})
+	assert.NilError(t, err)
+
+	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
+		}
+	}()
+
+	// Let it run for a while without incrementing the block number
+	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):
+		// Test passes if no error is received
+	}
+
+	// Verify the block number hasn't changed
+	syncedData, err := db.GetIdentityRegisteredEventsSyncedUntil(ctx)
+	assert.NilError(t, err)
+	assert.Equal(t, initialBlockNumber, syncedData.BlockNumber, "block number should remain unchanged")
+}
+
+func TestAPISyncMonitor_ContinuesWhenNoBatchConfig(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)
+
+	// Only set up initial block data, no keyper set
+	initialBlockNumber := int64(100)
+	err := db.SetIdentityRegisteredEventSyncedUntil(ctx, database.SetIdentityRegisteredEventSyncedUntilParams{
+		BlockHash:   []byte{0x01, 0x02, 0x03},
+		BlockNumber: initialBlockNumber,
+	})
+	assert.NilError(t, err)
+
+	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
+		}
+	}()
+
+	// Let it run for a while without incrementing the block number
+	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):
+		// Test passes if no error is received
+	}
+
+	// Verify the block number hasn't changed
+	syncedData, err := db.GetIdentityRegisteredEventsSyncedUntil(ctx)
+	assert.NilError(t, err)
+	assert.Equal(t, initialBlockNumber, syncedData.BlockNumber, "block number should remain unchanged")
+}