update sync monitor to use DKGStartBlockDelta to determine if dkg started

Author: blockchainluffy

rolling-shutter/keyperimpl/shutterservice/keyper.go

@@ -114,8 +114,9 @@ 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,
+		DBPool:             kpr.dbpool,
+		CheckInterval:      time.Duration(kpr.config.Chain.SyncMonitorCheckInterval) * time.Second,
+		DKGStartBlockDelta: kpr.config.Shuttermint.DKGStartBlockDelta,
 	}
 	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,8 +16,9 @@ import (
 )
 
 type SyncMonitor struct {
-	DBPool        *pgxpool.Pool
-	CheckInterval time.Duration
+	DBPool             *pgxpool.Pool
+	CheckInterval      time.Duration
+	DKGStartBlockDelta uint64
 }
 
 func (s *SyncMonitor) Start(ctx context.Context, runner service.Runner) error {
@@ -59,14 +60,9 @@ func (s *SyncMonitor) runCheck(
 	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)
+	if s.dkgIsRunning(ctx, keyperdb) {
+		log.Debug().Msg("dkg is running, skipping sync monitor checks")
+		return nil
 	}
 
 	record, err := db.GetIdentityRegisteredEventsSyncedUntil(ctx)
@@ -75,7 +71,7 @@ func (s *SyncMonitor) runCheck(
 			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")
+		return fmt.Errorf("error getting identity_registered_events_synced_until: %w", err)
 	}
 
 	currentBlockNumber := record.BlockNumber
@@ -86,13 +82,37 @@ func (s *SyncMonitor) runCheck(
 		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
+}
+
+func (s *SyncMonitor) dkgIsRunning(ctx context.Context, keyperdb *keyperDB.Queries) bool {
+	batchConfig, err := keyperdb.GetLatestBatchConfig(ctx)
+	if err != nil {
 		log.Error().
-			Int64("last-block-number", *lastBlockNumber).
-			Int64("current-block-number", currentBlockNumber).
-			Msg("block number has not increased between checks")
-		return ErrBlockNotIncreasing
+			Err(err).
+			Msg("syncMonitor | error getting latest batchconfig")
+		return true
 	}
 
-	return nil
+	tmSyncData, err := keyperdb.TMGetSyncMeta(ctx)
+	if err != nil {
+		log.Error().
+			Err(err).
+			Msg("syncMonitor | error getting TMSyncMeta")
+		return true
+	}
+
+	// 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
+		_, err := keyperdb.GetDKGResult(ctx, int64(batchConfig.KeyperConfigIndex))
+		if err != nil {
+			return true
+		}
+	}
+	return false
 }

rolling-shutter/keyperimpl/shutterservice/syncmonitor_test.go

@@ -24,6 +24,7 @@ func setupTestData(ctx context.Context, t *testing.T, dbpool *pgxpool.Pool, bloc
 	err := keyperdb.InsertBatchConfig(ctx, keyperDB.InsertBatchConfigParams{
 		KeyperConfigIndex: 1,
 		Keypers:           []string{},
+		Height:            50,
 	})
 	assert.NilError(t, err)
 
@@ -34,6 +35,12 @@ 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},
@@ -183,6 +190,13 @@ func TestAPISyncMonitor_ContinuesWhenNoDKGResult(t *testing.T) {
 	err := keyperdb.InsertBatchConfig(ctx, keyperDB.InsertBatchConfigParams{
 		KeyperConfigIndex: 1,
 		Keypers:           []string{},
+		Height:            50,
+	})
+	assert.NilError(t, err)
+
+	// Set up TMSyncMeta
+	err = keyperdb.TMSetSyncMeta(ctx, keyperDB.TMSetSyncMetaParams{
+		LastCommittedHeight: 100,
 	})
 	assert.NilError(t, err)
 
@@ -275,3 +289,73 @@ func TestAPISyncMonitor_ContinuesWhenNoBatchConfig(t *testing.T) {
 	assert.NilError(t, err)
 	assert.Equal(t, initialBlockNumber, syncedData.BlockNumber, "block number should remain unchanged")
 }
+
+func TestAPISyncMonitor_ContinuesWhenNoTMSyncMeta(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)
+
+	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
+	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,
+	})
+	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,
+		DKGStartBlockDelta: 5,
+	}
+
+	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
+	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")
+}