test: fix test Lazy aggregator, fast da sync, two rollups same namespace

block/batch.go

@@ -6,14 +6,16 @@ import (
 
 // BatchQueue is a queue of transaction batches with timestamps
 type BatchQueue struct {
-	queue []BatchWithTime
-	mu sync.Mutex
+	queue []BatchWithTime
+	mu sync.Mutex
+	notifyCh chan struct{}
 }
 
 // NewBatchQueue creates a new BatchQueue
 func NewBatchQueue() *BatchQueue {
 	return &BatchQueue{
- queue: make([]BatchWithTime, 0),
+ queue: make([]BatchWithTime, 0),
+ notifyCh: make(chan struct{}, 1),
 	}
 }
 
@@ -22,6 +24,11 @@ func (bq *BatchQueue) AddBatch(batch BatchWithTime) {
 	bq.mu.Lock()
 	defer bq.mu.Unlock()
 	bq.queue = append(bq.queue, batch)
+	select {
+	case bq.notifyCh <- struct{}{}: // Send notification if there's no pending notification
+	default:
+ // Do nothing if a notification is already pending
+	}
 }
 
 // Next returns the next batch in the queue

block/manager.go

@@ -78,6 +78,9 @@ var initialBackoff = 100 * time.Millisecond
 // DAIncludedHeightKey is the key used for persisting the da included height in store.
 const DAIncludedHeightKey = "da included height"
 
+// dataHashForEmptyTxs to be used while only syncing headers from DA and no p2p to get the Data for no txs scenarios, the syncing can proceed without getting stuck forever.
+var dataHashForEmptyTxs = []byte{110, 52, 11, 156, 255, 179, 122, 152, 156, 165, 68, 230, 187, 120, 10, 44, 120, 144, 29, 63, 179, 55, 56, 118, 133, 17, 163, 6, 23, 175, 160, 29}
+
 // NewHeaderEvent is used to pass header and DA height to headerInCh
 type NewHeaderEvent struct {
 	Header *types.SignedHeader
@@ -139,7 +142,6 @@ type Manager struct {
 
 	// For usage by Lazy Aggregator mode
 	buildingBlock bool
-	txsAvailable <-chan struct{}
 
 	pendingHeaders *PendingHeaders
 
@@ -261,13 +263,6 @@ func NewManager(
  return nil, err
 	}
 
-	var txsAvailableCh <-chan struct{}
-	if mempool != nil {
- txsAvailableCh = mempool.TxsAvailable()
-	} else {
- txsAvailableCh = nil
-	}
-
 	pendingHeaders, err := NewPendingHeaders(store, logger)
 	if err != nil {
  return nil, err
@@ -296,7 +291,6 @@ func NewManager(
  dataCache: NewDataCache(),
  retrieveCh: make(chan struct{}, 1),
  logger: logger,
- txsAvailable: txsAvailableCh,
  buildingBlock: false,
  pendingHeaders: pendingHeaders,
  metrics: seqMetrics,
@@ -497,10 +491,8 @@ func (m *Manager) lazyAggregationLoop(ctx context.Context, blockTimer *time.Time
  select {
  case <-ctx.Done():
  return
- // the txsAvailable channel is signalled when Txns become available
- // in the mempool, or after transactions remain in the mempool after
- // building a block.
- case _, ok := <-m.txsAvailable:
+ // the m.bq.notifyCh channel is signalled when batch becomes available in the batch queue
+ case _, ok := <-m.bq.notifyCh:
  if ok && !m.buildingBlock {
  // set the buildingBlock flag to prevent multiple calls to reset the time
  m.buildingBlock = true
@@ -563,6 +555,36 @@ func (m *Manager) HeaderSubmissionLoop(ctx context.Context) {
 	}
 }
 
+func (m *Manager) handleEmptyDataHash(ctx context.Context, header *types.Header) {
+	headerHeight := header.Height()
+	if bytes.Equal(header.DataHash, dataHashForEmptyTxs) {
+ var lastDataHash types.Hash
+ var err error
+ var lastData *types.Data
+ if headerHeight > 1 {
+ _, lastData, err = m.store.GetBlockData(ctx, headerHeight-1)
+ if lastData != nil {
+ lastDataHash = lastData.Hash()
+ }
+ }
+ // if err then we cannot populate data, hence just skip and wait for Data to be synced
+ if err == nil {
+ metadata := &types.Metadata{
+ ChainID: header.ChainID(),
+ Height: headerHeight,
+ Time: header.BaseHeader.Time,
+ LastDataHash: lastDataHash,
+ }
+ d := &types.Data{
+ Metadata: metadata,
+ }
+ m.dataCache.setData(headerHeight, d)
+ } else {
+ m.logger.Error("failed to get block data for", "height", headerHeight-1, "error", err)
+ }
+	}
+}
+
 // SyncLoop is responsible for syncing blocks.
 //
 // SyncLoop processes headers gossiped in P2P network to know what's the latest block height,
@@ -599,6 +621,11 @@ func (m *Manager) SyncLoop(ctx context.Context, cancel context.CancelFunc) {
  m.sendNonBlockingSignalToHeaderStoreCh()
  m.sendNonBlockingSignalToRetrieveCh()
 
+ // check if the dataHash is dataHashForEmptyTxs
+ // no need to wait for syncing Data, instead prepare now and set
+ // so that trySyncNextBlock can progress
+ m.handleEmptyDataHash(ctx, &header.Header)
+
  err := m.trySyncNextBlock(ctx, daHeight)
  if err != nil {
  m.logger.Info("failed to sync next block", "error", err)

block/manager_test.go

@@ -130,6 +130,52 @@ func TestInitialStateStored(t *testing.T) {
 	require.Equal(s.InitialHeight, uint64(1))
 }
 
+func TestHandleEmptyDataHash(t *testing.T) {
+	require := require.New(t)
+	ctx := context.Background()
+
+	// Mock store and data cache
+	store := mocks.NewStore(t)
+	dataCache := NewDataCache()
+
+	// Setup the manager with the mock and data cache
+	m := &Manager{
+ store: store,
+ dataCache: dataCache,
+	}
+
+	// Define the test data
+	headerHeight := 2
+	header := &types.Header{
+ DataHash: dataHashForEmptyTxs,
+ BaseHeader: types.BaseHeader{
+ Height: 2,
+ Time: uint64(time.Now().UnixNano()),
+ },
+	}
+
+	// Mock data for the previous block
+	lastData := &types.Data{}
+	lastDataHash := lastData.Hash()
+
+	// header.DataHash equals dataHashForEmptyTxs and no error occurs
+	store.On("GetBlockData", ctx, uint64(headerHeight-1)).Return(nil, lastData, nil)
+
+	// Execute the method under test
+	m.handleEmptyDataHash(ctx, header)
+
+	// Assertions
+	store.AssertExpectations(t)
+
+	// make sure that the store has the correct data
+	d := dataCache.getData(header.Height())
+	require.NotNil(d)
+	require.Equal(d.Metadata.LastDataHash, lastDataHash)
+	require.Equal(d.Metadata.ChainID, header.ChainID())
+	require.Equal(d.Metadata.Height, header.Height())
+	require.Equal(d.Metadata.Time, header.BaseHeader.Time)
+}
+
 func TestInitialStateUnexpectedHigherGenesis(t *testing.T) {
 	require := require.New(t)
 	genesisDoc, _ := types.GetGenesisWithPrivkey(types.DefaultSigningKeyType)
@@ -810,6 +856,7 @@ func TestAggregationLoop(t *testing.T) {
  BlockTime: time.Second,
  LazyAggregator: false,
  },
+ bq: NewBatchQueue(),
 	}
 
 	mockStore.On("Height").Return(uint64(0))
@@ -829,14 +876,13 @@ func TestAggregationLoop(t *testing.T) {
 func TestLazyAggregationLoop(t *testing.T) {
 	mockLogger := new(test.MockLogger)
 
-	txsAvailable := make(chan struct{}, 1)
 	m := &Manager{
- logger: mockLogger,
- txsAvailable: txsAvailable,
+ logger: mockLogger,
  conf: config.BlockManagerConfig{
  BlockTime: time.Second,
  LazyAggregator: true,
  },
+ bq: NewBatchQueue(),
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
@@ -846,7 +892,7 @@ func TestLazyAggregationLoop(t *testing.T) {
 	defer blockTimer.Stop()
 
 	go m.lazyAggregationLoop(ctx, blockTimer)
-txsAvailable <- struct{}{}
+m.bq.notifyCh <- struct{}{}
 
 	// Wait for the function to complete or timeout
 	<-ctx.Done()