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

update sync monitor to not check for dkg and ws error to panic keyper

Author: blockchainluffy

rolling-shutter/keyperimpl/shutterservice/syncmonitor.go

@@ -59,15 +59,6 @@ func (s *SyncMonitor) runCheck(
 	keyperdb *keyperDB.Queries,
 	lastBlockNumber *int64,
 ) error {
-	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
-	}
-
 	record, err := db.GetIdentityRegisteredEventsSyncedUntil(ctx)
 	if err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
@@ -91,31 +82,3 @@ func (s *SyncMonitor) runCheck(
 		Msg("block number has not increased between checks")
 	return ErrBlockNotIncreasing
 }
-
-func (s *SyncMonitor) isDKGRunning(ctx context.Context, keyperdb *keyperDB.Queries) (bool, error) {
-	// if latest eon is registered then EonStarted event has triggered, which means the dkg can start
-	eons, err := keyperdb.GetAllEons(ctx)
-	if errors.Is(err, pgx.ErrNoRows) {
-		return false, nil
-	}
-	if err != nil {
-		log.Error().
-			Err(err).
-			Msg("syncMonitor | error getting all eons")
-		return false, err
-	}
-
-	if len(eons) == 0 {
-		return false, nil
-	}
-
-	// if we get no rows in getting dkg result then dkg is not completed for that eon
-	_, err = keyperdb.GetDKGResult(ctx, eons[len(eons)-1].Eon)
-	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, nil
-}

rolling-shutter/keyperimpl/shutterservice/syncmonitor_test.go

@@ -8,7 +8,6 @@ import (
 	"github.com/jackc/pgx/v4/pgxpool"
 	"gotest.tools/assert"
 
-	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"
@@ -18,23 +17,9 @@ import (
 func setupTestData(ctx context.Context, t *testing.T, dbpool *pgxpool.Pool, blockNumber int64) {
 	t.Helper()
 	db := database.New(dbpool)
-	keyperdb := keyperDB.New(dbpool)
-
-	// Set up eon
-	err := keyperdb.InsertEon(ctx, keyperDB.InsertEonParams{
-		Eon: 1,
-	})
-	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{
+	err := db.SetIdentityRegisteredEventSyncedUntil(ctx, database.SetIdentityRegisteredEventSyncedUntilParams{
 		BlockHash:   []byte{0x01, 0x02, 0x03},
 		BlockNumber: blockNumber,
 	})
@@ -121,132 +106,13 @@ func TestAPISyncMonitor_HandlesBlockNumberIncreasing(t *testing.T) {
 	assert.Equal(t, initialBlockNumber+5, syncedData.BlockNumber, "block number should have been incremented correctly")
 }
 
-func TestAPISyncMonitor_SkipsWhenDKGIsRunning(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 eon but no DKG result to simulate DKG running
-	err := keyperdb.InsertEon(ctx, keyperDB.InsertEonParams{
-		Eon: 1,
-	})
-	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_RunsNormallyWhenNoEons(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 eon entries
-	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:
-		assert.ErrorContains(t, err, shutterservice.ErrBlockNotIncreasing.Error())
-	case <-time.After(1 * time.Second):
-		t.Fatalf("expected monitor to throw error, but no error returned")
-	}
-
-	// 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_ContinuesWhenNoRows(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
 	dbpool, closeDB := testsetup.NewTestDBPool(ctx, t, database.Definition)
 	defer closeDB()
 
-	// Set up eon and DKG result, but no block data
-	keyperdb := keyperDB.New(dbpool)
-
-	err := keyperdb.InsertEon(ctx, keyperDB.InsertEonParams{
-		Eon: 1,
-	})
-	assert.NilError(t, err)
-
-	err = keyperdb.InsertDKGResult(ctx, keyperDB.InsertDKGResultParams{
-		Eon:     1,
-		Success: true,
-	})
-	assert.NilError(t, err)
-
 	monitor := &shutterservice.SyncMonitor{
 		DBPool:        dbpool,
 		CheckInterval: 5 * time.Second,
@@ -272,54 +138,3 @@ func TestAPISyncMonitor_ContinuesWhenNoRows(t *testing.T) {
 	case <-time.After(1 * time.Second):
 	}
 }
-
-func TestAPISyncMonitor_RunsNormallyWithCompletedDKG(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)
-	setupTestData(ctx, t, dbpool, initialBlockNumber)
-
-	// Set up initial block data
-	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:
-		assert.ErrorContains(t, err, shutterservice.ErrBlockNotIncreasing.Error())
-	case <-time.After(1 * time.Second):
-		t.Fatalf("expected monitor to throw error, but no error returned")
-	}
-
-	// 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")
-}

rolling-shutter/medley/chainsync/syncer/eonpubkey.go

@@ -149,7 +149,8 @@ func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context, subsErr <-chan
 			}
 		case err := <-subsErr:
 			if err != nil {
-				s.Log.Error("subscription error for watchNewEonPubkey", err)
+				s.Log.Error("subscription error for watchNewEonPubkey", err.Error())
+				return err
 			}
 		case <-ctx.Done():
 			return ctx.Err()

rolling-shutter/medley/chainsync/syncer/keyperset.go

@@ -236,7 +236,8 @@ func (s *KeyperSetSyncer) watchNewKeypersService(ctx context.Context, subsErr <-
 			}
 		case err := <-subsErr:
 			if err != nil {
-				s.Log.Error("subscription error for watchNewKeypersService", err)
+				s.Log.Error("subscription error for watchNewKeypersService", err.Error())
+				return err
 			}
 		case <-ctx.Done():
 			return ctx.Err()

rolling-shutter/medley/chainsync/syncer/shutterstate.go

@@ -125,11 +125,13 @@ func (s *ShutterStateSyncer) watchPaused(ctx context.Context, subsErr <-chan err
 			s.handle(ctx, ev)
 		case err := <-subsErr:
 			if err != nil {
-				s.Log.Error("subscription error for watchPaused")
+				s.Log.Error("subscription error for watchPaused", err.Error())
+				return err
 			}
 		case err := <-subsErrUnpaused:
 			if err != nil {
-				s.Log.Error("subscription error for watchUnpaused", err)
+				s.Log.Error("subscription error for watchUnpaused", err.Error())
+				return err
 			}
 		case <-ctx.Done():
 			return ctx.Err()

rolling-shutter/medley/chainsync/syncer/unsafehead.go

@@ -60,7 +60,8 @@ func (s *UnsafeHeadSyncer) watchLatestUnsafeHead(ctx context.Context, subsErr <-
 			}
 		case err := <-subsErr:
 			if err != nil {
-				s.Log.Error("subscription error for watchLatestUnsafeHead", err)
+				s.Log.Error("subscription error for watchLatestUnsafeHead", err.Error())
+				return err
 			}
 		case <-ctx.Done():
 			return ctx.Err()