updated sync monitor to check for BatchConfigStarted event for DKG

Author: blockchainluffy

rolling-shutter/keyperimpl/shutterservice/keyper.go

@@ -114,9 +114,8 @@ func (kpr *Keyper) Start(ctx context.Context, runner service.Runner) error {
 	}
 
 	kpr.syncMonitor = &SyncMonitor{
-		DBPool:             kpr.dbpool,
-		CheckInterval:      time.Duration(kpr.config.Chain.SyncMonitorCheckInterval) * time.Second,
-		DKGStartBlockDelta: kpr.config.Shuttermint.DKGStartBlockDelta,
+		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, kpr.syncMonitor)

rolling-shutter/keyperimpl/shutterservice/syncmonitor.go

@@ -16,9 +16,8 @@ import (
 )
 
 type SyncMonitor struct {
-	DBPool             *pgxpool.Pool
-	CheckInterval      time.Duration
-	DKGStartBlockDelta uint64
+	DBPool        *pgxpool.Pool
+	CheckInterval time.Duration
 }
 
 func (s *SyncMonitor) Start(ctx context.Context, runner service.Runner) error {
@@ -60,7 +59,11 @@ func (s *SyncMonitor) runCheck(
 	keyperdb *keyperDB.Queries,
 	lastBlockNumber *int64,
 ) error {
-	if s.dkgIsRunning(ctx, keyperdb) {
+	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
 	}
@@ -89,30 +92,28 @@ func (s *SyncMonitor) runCheck(
 	return ErrBlockNotIncreasing
 }
 
-func (s *SyncMonitor) dkgIsRunning(ctx context.Context, keyperdb *keyperDB.Queries) bool {
+func (s *SyncMonitor) isDKGRunning(ctx context.Context, keyperdb *keyperDB.Queries) (bool, error) {
 	batchConfig, err := keyperdb.GetLatestBatchConfig(ctx)
-	if err != nil {
-		log.Error().
-			Err(err).
-			Msg("syncMonitor | error getting latest batchconfig")
-		return true
+	if errors.Is(err, pgx.ErrNoRows) {
+		return false, nil
 	}
-
-	tmSyncData, err := keyperdb.TMGetSyncMeta(ctx)
 	if err != nil {
 		log.Error().
 			Err(err).
-			Msg("syncMonitor | error getting TMSyncMeta")
-		return true
+			Msg("syncMonitor | error getting latest batchconfig")
+		return false, err
 	}
 
-	// if batchConfig submission height + DKGStartBlockDelta is greater than shuttermint height then dkg has not started yet
-	if batchConfig.Height+int64(s.DKGStartBlockDelta) < tmSyncData.LastCommittedHeight {
-		// if we get an error in getting dkg result then dkg is not completed
+	// if batchconfig.Started is true that means dkg can happen for this keyper config index
+	if batchConfig.Started {
+		// if we get no rows in getting dkg result then dkg is not completed for that keyper config index
 		_, err := keyperdb.GetDKGResult(ctx, int64(batchConfig.KeyperConfigIndex))
-		if err != nil {
-			return true
+		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
+	return false, nil
 }

rolling-shutter/keyperimpl/shutterservice/syncmonitor_test.go

@@ -20,11 +20,12 @@ func setupTestData(ctx context.Context, t *testing.T, dbpool *pgxpool.Pool, bloc
 	db := database.New(dbpool)
 	keyperdb := keyperDB.New(dbpool)
 
-	// Set up batch config
+	// Set up batch config with Started flag
 	err := keyperdb.InsertBatchConfig(ctx, keyperDB.InsertBatchConfigParams{
 		KeyperConfigIndex: 1,
 		Keypers:           []string{},
 		Height:            50,
+		Started:           true,
 	})
 	assert.NilError(t, err)
 
@@ -35,12 +36,6 @@ func setupTestData(ctx context.Context, t *testing.T, dbpool *pgxpool.Pool, bloc
 	})
 	assert.NilError(t, err)
 
-	// Set up TMSyncMeta
-	err = keyperdb.TMSetSyncMeta(ctx, keyperDB.TMSetSyncMetaParams{
-		LastCommittedHeight: 100,
-	})
-	assert.NilError(t, err)
-
 	// Set up initial block
 	err = db.SetIdentityRegisteredEventSyncedUntil(ctx, database.SetIdentityRegisteredEventSyncedUntilParams{
 		BlockHash:   []byte{0x01, 0x02, 0x03},
@@ -142,6 +137,7 @@ func TestAPISyncMonitor_ContinuesWhenNoRows(t *testing.T) {
 	err := keyperdb.InsertBatchConfig(ctx, keyperDB.InsertBatchConfigParams{
 		KeyperConfigIndex: 1,
 		Keypers:           []string{},
+		Started:           true,
 	})
 	assert.NilError(t, err)
 
@@ -177,7 +173,7 @@ func TestAPISyncMonitor_ContinuesWhenNoRows(t *testing.T) {
 	}
 }
 
-func TestAPISyncMonitor_ContinuesWhenNoDKGResult(t *testing.T) {
+func TestAPISyncMonitor_SkipsWhenDKGIsRunning(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -191,12 +187,7 @@ func TestAPISyncMonitor_ContinuesWhenNoDKGResult(t *testing.T) {
 		KeyperConfigIndex: 1,
 		Keypers:           []string{},
 		Height:            50,
-	})
-	assert.NilError(t, err)
-
-	// Set up TMSyncMeta
-	err = keyperdb.TMSetSyncMeta(ctx, keyperDB.TMSetSyncMetaParams{
-		LastCommittedHeight: 100,
+		Started:           true,
 	})
 	assert.NilError(t, err)
 
@@ -241,7 +232,7 @@ func TestAPISyncMonitor_ContinuesWhenNoDKGResult(t *testing.T) {
 	assert.Equal(t, initialBlockNumber, syncedData.BlockNumber, "block number should remain unchanged")
 }
 
-func TestAPISyncMonitor_ContinuesWhenNoBatchConfig(t *testing.T) {
+func TestAPISyncMonitor_RunsNormallyWhenNoBatchConfig(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -279,9 +270,9 @@ func TestAPISyncMonitor_ContinuesWhenNoBatchConfig(t *testing.T) {
 
 	select {
 	case err := <-errCh:
-		t.Fatalf("expected monitor to continue without error, but got: %v", err)
+		assert.ErrorContains(t, err, shutterservice.ErrBlockNotIncreasing.Error())
 	case <-time.After(1 * time.Second):
-		// Test passes if no error is received
+		t.Fatalf("expected monitor to throw error, but no error returned")
 	}
 
 	// Verify the block number hasn't changed
@@ -290,7 +281,7 @@ func TestAPISyncMonitor_ContinuesWhenNoBatchConfig(t *testing.T) {
 	assert.Equal(t, initialBlockNumber, syncedData.BlockNumber, "block number should remain unchanged")
 }
 
-func TestAPISyncMonitor_ContinuesWhenNoTMSyncMeta(t *testing.T) {
+func TestAPISyncMonitor_RunsNormallyWhenBatchConfigNotStarted(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -299,22 +290,12 @@ func TestAPISyncMonitor_ContinuesWhenNoTMSyncMeta(t *testing.T) {
 	db := database.New(dbpool)
 	keyperdb := keyperDB.New(dbpool)
 
-	batchConfigHeight := int64(50)
-	// Set TMSyncData height to be less than batchConfigHeight + DKGStartBlockDelta
-	// This simulates a scenario where DKG hasn't started yet
-	tmSyncHeight := int64(60)
-
-	// Set up batch config
+	// Set up batch config with Started = false
 	err := keyperdb.InsertBatchConfig(ctx, keyperDB.InsertBatchConfigParams{
 		KeyperConfigIndex: 1,
 		Keypers:           []string{},
-		Height:            batchConfigHeight,
-	})
-	assert.NilError(t, err)
-
-	// Set up TMSyncMeta with lower height
-	err = keyperdb.TMSetSyncMeta(ctx, keyperDB.TMSetSyncMetaParams{
-		LastCommittedHeight: tmSyncHeight,
+		Height:            50,
+		Started:           false,
 	})
 	assert.NilError(t, err)
 
@@ -327,9 +308,8 @@ func TestAPISyncMonitor_ContinuesWhenNoTMSyncMeta(t *testing.T) {
 	assert.NilError(t, err)
 
 	monitor := &shutterservice.SyncMonitor{
-		DBPool:             dbpool,
-		CheckInterval:      5 * time.Second,
-		DKGStartBlockDelta: 5,
+		DBPool:        dbpool,
+		CheckInterval: 5 * time.Second,
 	}
 
 	monitorCtx, cancelMonitor := context.WithCancel(ctx)
@@ -349,9 +329,9 @@ func TestAPISyncMonitor_ContinuesWhenNoTMSyncMeta(t *testing.T) {
 
 	select {
 	case err := <-errCh:
-		t.Fatalf("expected monitor to continue without error, but got: %v", err)
+		assert.ErrorContains(t, err, shutterservice.ErrBlockNotIncreasing.Error())
 	case <-time.After(1 * time.Second):
-		// Test passes if no error is received
+		t.Fatalf("expected monitor to throw error, but no error returned")
 	}
 
 	// Verify the block number hasn't changed