Merge remote-tracking branch 'origin/develop' into issue-1198

Author: rachit77

aggsender/block_notifier_polling.go

@@ -40,28 +40,28 @@ type BlockNotifierPolling struct {
 	config        ConfigBlockNotifierPolling
 	mu            sync.Mutex
 	lastStatus    *blockNotifierPollingInternalStatus
-	types.GenericSubscriber[types.EventNewBlock]
+	aggkitcommon.PubSub[types.EventNewBlock]
 }
 
 // NewBlockNotifierPolling creates a new BlockNotifierPolling.
-// if param `subscriber` is nil a new GenericSubscriberImpl[types.EventNewBlock] will be created.
+// if param `subscriber` is nil a new GenericSubscriber[types.EventNewBlock] will be created.
 // To use this class you need to subscribe and each time that a new block appear the subscriber
 // will be notified through the channel. (check unit tests TestExploratoryBlockNotifierPolling
 // for more information)
 func NewBlockNotifierPolling(ethClient aggkittypes.BaseEthereumClienter,
 	config ConfigBlockNotifierPolling,
 	logger aggkitcommon.Logger,
-	subscriber types.GenericSubscriber[types.EventNewBlock]) (*BlockNotifierPolling, error) {
+	subscriber aggkitcommon.PubSub[types.EventNewBlock]) (*BlockNotifierPolling, error) {
 	if subscriber == nil {
-		subscriber = NewGenericSubscriberImpl[types.EventNewBlock]()
+		subscriber = aggkitcommon.NewGenericSubscriber[types.EventNewBlock]()
 	}
 
 	return &BlockNotifierPolling{
-		ethClient:         ethClient,
-		blockFinality:     config.BlockFinalityType,
-		logger:            logger,
-		config:            config,
-		GenericSubscriber: subscriber,
+		ethClient:     ethClient,
+		blockFinality: config.BlockFinalityType,
+		logger:        logger,
+		config:        config,
+		PubSub:        subscriber,
 	}, nil
 }
 

aggsender/epoch_notifier_per_block.go

@@ -74,15 +74,15 @@ type EpochNotifierPerBlock struct {
 	lastStartingEpochBlock uint64
 
 	Config ConfigEpochNotifierPerBlock
-	types.GenericSubscriber[types.EpochEvent]
+	aggkitcommon.PubSub[types.EpochEvent]
 }
 
 func NewEpochNotifierPerBlock(blockNotifier types.BlockNotifier,
 	logger aggkitcommon.Logger,
 	config ConfigEpochNotifierPerBlock,
-	subscriber types.GenericSubscriber[types.EpochEvent]) (*EpochNotifierPerBlock, error) {
+	subscriber aggkitcommon.PubSub[types.EpochEvent]) (*EpochNotifierPerBlock, error) {
 	if subscriber == nil {
-		subscriber = NewGenericSubscriberImpl[types.EpochEvent]()
+		subscriber = aggkitcommon.NewGenericSubscriber[types.EpochEvent]()
 	}
 
 	err := config.Validate()
@@ -94,7 +94,7 @@ func NewEpochNotifierPerBlock(blockNotifier types.BlockNotifier,
 		logger:                 logger,
 		lastStartingEpochBlock: config.StartingEpochBlock,
 		Config:                 config,
-		GenericSubscriber:      subscriber,
+		PubSub:                 subscriber,
 	}, nil
 }
 

aggsender/generic_subscriber_impl.go

@@ -784,7 +784,7 @@ func (b *BridgeService) InjectedL1InfoLeafHandler(c *gin.Context) {
 	}
 
 	if err != nil {
-		b.logger.Errorf("failed to get L1 info tree leaf (network id=%d, leaf index=%d): %v", networkID, l1InfoTreeIndex, err)
+		b.logger.Debugf("failed to get L1 info tree leaf (network id=%d, leaf index=%d): %v", networkID, l1InfoTreeIndex, err)
 		c.JSON(http.StatusInternalServerError,
 			gin.H{"error": fmt.Sprintf("failed to get L1 info tree leaf (network id=%d, leaf index=%d), error: %s",
 				networkID, l1InfoTreeIndex, err)})

aggsender/mocks/mock_generic_subscriber.go

@@ -70,7 +70,6 @@ func NewL1(
 	rd ReorgDetector,
 	ethClient aggkittypes.EthClienter,
 	originNetwork uint32,
-	syncFullClaims bool,
 ) (*BridgeSync, error) {
 	return newBridgeSync(
 		ctx,
@@ -80,7 +79,7 @@ func NewL1(
 		ethClient,
 		L1BridgeSyncer,
 		originNetwork,
-		syncFullClaims,
+		false,
 	)
 }
 
@@ -405,6 +404,11 @@ func (s *BridgeSync) OriginNetwork() uint32 {
 	return s.originNetwork
 }
 
+// SubscribeToSync allows a subscriber to receive block notifications
+func (s *BridgeSync) SubscribeToSync(subscriberID string) <-chan sync.Block {
+	return s.driver.SubscribeToNewBlocks(subscriberID)
+}
+
 type LastReorg struct {
 	DetectedAt int64  `json:"detected_at"`
 	FromBlock  uint64 `json:"from_block"`

aggsender/types/generic_subscriber.go

@@ -75,7 +75,6 @@ func TestNewLx(t *testing.T) {
 		mockReorgDetector,
 		mockEthClient,
 		originNetwork,
-		false,
 	)
 
 	require.NoError(t, err)
@@ -766,3 +765,111 @@ func TestBridgeSync_GetLastRoot(t *testing.T) {
 		require.NotEqual(t, common.Hash{}, root.Hash)
 	})
 }
+
+func TestBridgeSync_SubscribeToSync(t *testing.T) {
+	const (
+		syncBlockChunkSize         = uint64(100)
+		initialBlock               = uint64(0)
+		waitForNewBlocksPeriod     = time.Second * 10
+		retryAfterErrorPeriod      = time.Second * 5
+		maxRetryAttemptsAfterError = 3
+		originNetwork              = uint32(1)
+	)
+
+	var (
+		blockFinalityType = aggkittypes.SafeBlock
+		ctx               = context.Background()
+		dbPath            = path.Join(t.TempDir(), "TestSubscribeToSync.sqlite")
+		bridge            = common.HexToAddress("0x1234567890abcdef1234567890abcdef12345678")
+	)
+
+	mockEthClient := mocksethclient.NewEthClienter(t)
+	mockEthClient.EXPECT().CallContract(mock.Anything, mock.Anything, mock.Anything).Return(
+		common.FromHex("0x000000000000000000000000000000000000000000000000000000000000002a"), nil).Once()
+	mockEthClient.EXPECT().
+		CallContract(
+			mock.Anything,
+			mock.Anything,
+			mock.Anything,
+		).
+		Return(common.LeftPadBytes(common.HexToAddress("0x3c351e10").Bytes(), 32), nil).
+		Maybe()
+	mockReorgDetector := mocksbridgesync.NewReorgDetector(t)
+
+	mockReorgDetector.EXPECT().Subscribe(mock.Anything).Return(nil, nil)
+	mockReorgDetector.EXPECT().GetFinalizedBlockType().Return(blockFinalityType)
+	mockReorgDetector.EXPECT().String().Return("mockReorgDetector")
+
+	dbQueryTimeout := 30 * time.Second
+
+	bridgeSyncCfg := Config{
+		DBPath:                             dbPath,
+		BridgeAddr:                         bridge,
+		BlockFinality:                      aggkittypes.LatestBlock,
+		SyncBlockChunkSize:                 syncBlockChunkSize,
+		InitialBlockNum:                    initialBlock,
+		WaitForNewBlocksPeriod:             cfgtypes.NewDuration(waitForNewBlocksPeriod),
+		RetryAfterErrorPeriod:              cfgtypes.NewDuration(retryAfterErrorPeriod),
+		MaxRetryAttemptsAfterError:         maxRetryAttemptsAfterError,
+		RequireStorageContentCompatibility: false,
+		DBQueryTimeout:                     cfgtypes.NewDuration(dbQueryTimeout),
+	}
+
+	s, err := NewL2(
+		ctx,
+		bridgeSyncCfg,
+		mockReorgDetector,
+		mockEthClient,
+		originNetwork,
+		false,
+	)
+	require.NoError(t, err)
+
+	t.Run("subscribe to sync with valid parameters", func(t *testing.T) {
+		subscriberID := "test-subscriber"
+
+		blockChan := s.SubscribeToSync(subscriberID)
+		require.NotNil(t, blockChan)
+
+		// Verify the channel is not closed immediately
+		select {
+		case _, ok := <-blockChan:
+			if !ok {
+				t.Fatal("channel should not be closed immediately")
+			}
+		default:
+			// Expected - no blocks available initially
+		}
+	})
+
+	t.Run("subscribe with empty subscriber ID", func(t *testing.T) {
+		subscriberID := ""
+
+		blockChan := s.SubscribeToSync(subscriberID)
+		require.NotNil(t, blockChan)
+	})
+
+	t.Run("multiple subscribers", func(t *testing.T) {
+		subscriber1ID := "subscriber-1"
+		subscriber2ID := "subscriber-2"
+
+		blockChan1 := s.SubscribeToSync(subscriber1ID)
+		blockChan2 := s.SubscribeToSync(subscriber2ID)
+
+		require.NotNil(t, blockChan1)
+		require.NotNil(t, blockChan2)
+
+		// Channels should be different instances
+		require.NotEqual(t, blockChan1, blockChan2)
+	})
+
+	t.Run("subscribe with same subscriber ID multiple times", func(t *testing.T) {
+		subscriberID := "duplicate-subscriber"
+
+		blockChan1 := s.SubscribeToSync(subscriberID)
+		blockChan2 := s.SubscribeToSync(subscriberID)
+
+		require.NotNil(t, blockChan1)
+		require.NotNil(t, blockChan2)
+	})
+}