test: Add back missing node integration tests

apps/evm/based/cmd/run.go

@@ -117,7 +117,7 @@ func NewExtendedRunNodeCmd(ctx context.Context) *cobra.Command {
  }
  rollDA = &client.DA
  } else {
- rollDA = coreda.NewDummyDA(100_000, 0, 0)
+ rollDA = coreda.NewDummyDA(100_000, 0, 0, nodeConfig.DA.BlockTime.Duration)
  }
 
  var basedDA coreda.DA
@@ -128,7 +128,7 @@ func NewExtendedRunNodeCmd(ctx context.Context) *cobra.Command {
  }
  basedDA = &client.DA
  } else {
- basedDA = coreda.NewDummyDA(100_000, 0, 0)
+ basedDA = coreda.NewDummyDA(100_000, 0, 0, nodeConfig.DA.BlockTime.Duration)
  }
 
  datastore, err := store.NewDefaultKVStore(nodeConfig.RootDir, nodeConfig.DBPath, "based")

block/aggregation.go

@@ -34,8 +34,8 @@ func (m *Manager) AggregationLoop(ctx context.Context, errCh chan<- error) {
 	blockTimer := time.NewTimer(0)
 	defer blockTimer.Stop()
 
-	// Lazy Aggregator mode.
-	// In Lazy Aggregator mode, blocks are built only when there are
+	// Lazy Sequencer mode.
+	// In Lazy Sequencer mode, blocks are built only when there are
 	// transactions or every LazyBlockTime.
 	if m.config.Node.LazyMode {
  if err := m.lazyAggregationLoop(ctx, blockTimer); err != nil {

block/da_includer_test.go

@@ -24,6 +24,8 @@ func newTestManager(t *testing.T) (*Manager, *mocks.Store, *mocks.Executor, *Moc
 	logger.On("Info", mock.Anything, mock.Anything).Maybe()
 	logger.On("Warn", mock.Anything, mock.Anything).Maybe()
 	logger.On("Error", mock.Anything, mock.Anything).Maybe()
+	// Mock Height to always return a high value so IsDAIncluded works
+	store.On("Height", mock.Anything).Return(uint64(100), nil).Maybe()
 	m := &Manager{
  store: store,
  headerCache: cache.NewCache[types.SignedHeader](),

block/manager.go

@@ -442,6 +442,13 @@ func (m *Manager) IsBlockHashSeen(blockHash string) bool {
 // IsDAIncluded returns true if the block with the given hash has been seen on DA.
 // TODO(tac0turtle): should we use this for pending header system to verify how far ahead a chain is?
 func (m *Manager) IsDAIncluded(ctx context.Context, height uint64) (bool, error) {
+	syncedHeight, err := m.store.Height(ctx)
+	if err != nil {
+ return false, err
+	}
+	if syncedHeight < height {
+ return false, nil
+	}
 	header, data, err := m.store.GetBlockData(ctx, height)
 	if err != nil {
  return false, err

block/manager_test.go

@@ -61,7 +61,7 @@ func getManager(t *testing.T, da da.DA, gasPrice float64, gasMultiplier float64)
 // TestInitialStateClean verifies that getInitialState initializes state correctly when no state is stored.
 func TestInitialStateClean(t *testing.T) {
 	require := require.New(t)
-	ctx := context.TODO()
+	ctx := t.Context()
 
 	// Create genesis document
 	genesisData, _, _ := types.GetGenesisWithPrivkey("TestInitialStateClean")
@@ -186,6 +186,7 @@ func TestIsDAIncluded(t *testing.T) {
 	height := uint64(1)
 	header, data := types.GetRandomBlock(height, 5, "TestIsDAIncluded")
 	mockStore.On("GetBlockData", mock.Anything, height).Return(header, data, nil).Times(3)
+	mockStore.On("Height", mock.Anything).Return(uint64(100), nil).Maybe()
 	ctx := context.Background()
 	// IsDAIncluded should return false for unseen hash
 	require.False(m.IsDAIncluded(ctx, height))

block/retriever.go

@@ -76,7 +76,7 @@ func (m *Manager) processNextDAHeaderAndData(ctx context.Context) error {
  m.logger.Debug("no blob data found", "daHeight", daHeight, "reason", blobsResp.Message)
  return nil
  }
- m.logger.Debug("retrieved potential data", "n", len(blobsResp.Data), "daHeight", daHeight)
+ m.logger.Debug("retrieved potential blob data", "n", len(blobsResp.Data), "daHeight", daHeight)
  for _, bz := range blobsResp.Data {
  if len(bz) == 0 {
  m.logger.Debug("ignoring nil or empty blob", "daHeight", daHeight)

block/sync.go

@@ -46,9 +46,6 @@ func (m *Manager) SyncLoop(ctx context.Context, errCh chan<- error) {
  }
  m.headerCache.SetItem(headerHeight, header)
 
- 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
@@ -70,12 +67,22 @@ func (m *Manager) SyncLoop(ctx context.Context, errCh chan<- error) {
  m.headerCache.SetSeen(headerHash)
  case dataEvent := <-m.dataInCh:
  data := dataEvent.Data
+ if len(data.Txs) == 0 {
+ continue
+ }
+
  daHeight := dataEvent.DAHeight
  dataHash := data.DACommitment().String()
+ dataHeight := uint64(0)
+ if data.Metadata != nil {
+ dataHeight = data.Metadata.Height
+ }
  m.logger.Debug("data retrieved",
  "daHeight", daHeight,
  "hash", dataHash,
+ "height", dataHeight,
  )
+
  if m.dataCache.IsSeen(dataHash) {
  m.logger.Debug("data already seen", "data hash", dataHash)
  continue
@@ -87,13 +94,12 @@ func (m *Manager) SyncLoop(ctx context.Context, errCh chan<- error) {
  }
  if data.Metadata != nil {
  // Data was sent via the P2P network
- dataHeight := data.Metadata.Height
+ dataHeight = data.Metadata.Height
  if dataHeight <= height {
  m.logger.Debug("data already seen", "height", dataHeight, "data hash", dataHash)
  continue
  }
  m.dataCache.SetItem(dataHeight, data)
- m.dataCache.SetItemByHash(dataHash, data)
  }
  // If the header is synced already, the data commitment should be associated with a height
  if val, ok := m.dataCommitmentToHeight.Load(dataHash); ok {
@@ -109,9 +115,6 @@ func (m *Manager) SyncLoop(ctx context.Context, errCh chan<- error) {
 
  m.dataCache.SetItemByHash(dataHash, data)
 
- m.sendNonBlockingSignalToDataStoreCh()
- m.sendNonBlockingSignalToRetrieveCh()
-
  err = m.trySyncNextBlock(ctx, daHeight)
  if err != nil {
  errCh <- fmt.Errorf("failed to sync next block: %w", err)
@@ -164,6 +167,10 @@ func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
  return fmt.Errorf("failed to apply block: %w", err)
  }
 
+ if err = m.updateState(ctx, newState); err != nil {
+ return fmt.Errorf("failed to save updated state: %w", err)
+ }
+
  if err = m.store.SaveBlockData(ctx, h, d, &h.Signature); err != nil {
  return fmt.Errorf("failed to save block: %w", err)
  }
@@ -177,10 +184,6 @@ func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
  newState.DAHeight = daHeight
  }
 
- if err = m.updateState(ctx, newState); err != nil {
- return fmt.Errorf("failed to save updated state: %w", err)
- }
-
  m.headerCache.DeleteItem(currentHeight + 1)
  m.dataCache.DeleteItem(currentHeight + 1)
  m.dataCache.DeleteItemByHash(h.DataHash.String())

core/da/dummy.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 )
@@ -19,10 +20,17 @@ type DummyDA struct {
 	maxBlobSize uint64
 	gasPrice float64
 	gasMultiplier float64
+
+	// DA height simulation
+	currentHeight uint64
+	blockTime time.Duration
+	stopCh chan struct{}
 }
 
-// NewDummyDA creates a new instance of DummyDA with the specified maximum blob size.
-func NewDummyDA(maxBlobSize uint64, gasPrice float64, gasMultiplier float64) *DummyDA {
+var ErrHeightFromFutureStr = fmt.Errorf("given height is from the future")
+
+// NewDummyDA creates a new instance of DummyDA with the specified maximum blob size and block time.
+func NewDummyDA(maxBlobSize uint64, gasPrice float64, gasMultiplier float64, blockTime time.Duration) *DummyDA {
 	return &DummyDA{
  blobs: make(map[string]Blob),
  commitments: make(map[string]Commitment),
@@ -32,9 +40,34 @@ func NewDummyDA(maxBlobSize uint64, gasPrice float64, gasMultiplier float64) *Du
  maxBlobSize: maxBlobSize,
  gasPrice: gasPrice,
  gasMultiplier: gasMultiplier,
+ blockTime: blockTime,
+ stopCh: make(chan struct{}),
 	}
 }
 
+// StartHeightTicker starts a goroutine that increments currentHeight every blockTime.
+func (d *DummyDA) StartHeightTicker() {
+	go func() {
+ ticker := time.NewTicker(d.blockTime)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ticker.C:
+ d.mu.Lock()
+ d.currentHeight++
+ d.mu.Unlock()
+ case <-d.stopCh:
+ return
+ }
+ }
+	}()
+}
+
+// StopHeightTicker stops the height ticker goroutine.
+func (d *DummyDA) StopHeightTicker() {
+	close(d.stopCh)
+}
+
 // MaxBlobSize returns the maximum blob size.
 func (d *DummyDA) MaxBlobSize(ctx context.Context) (uint64, error) {
 	return d.maxBlobSize, nil
@@ -71,6 +104,10 @@ func (d *DummyDA) GetIDs(ctx context.Context, height uint64, namespace []byte) (
 	d.mu.RLock()
 	defer d.mu.RUnlock()
 
+	if height > d.currentHeight {
+ return nil, fmt.Errorf("%w: requested %d, current %d", ErrHeightFromFutureStr, height, d.currentHeight)
+	}
+
 	ids, exists := d.blobsByHeight[height]
 	if !exists {
  return &GetIDsResult{
@@ -125,7 +162,7 @@ func (d *DummyDA) SubmitWithOptions(ctx context.Context, blobs []Blob, gasPrice
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
-	height := uint64(len(d.blobsByHeight))
+	height := d.currentHeight + 1
 	ids := make([]ID, 0, len(blobs))
 	var currentSize uint64
 
@@ -164,7 +201,12 @@ func (d *DummyDA) SubmitWithOptions(ctx context.Context, blobs []Blob, gasPrice
  ids = append(ids, id)
 	}
 
-	d.blobsByHeight[height] = ids
+	// Add the IDs to the blobsByHeight map if they don't already exist
+	if existingIDs, exists := d.blobsByHeight[height]; exists {
+ d.blobsByHeight[height] = append(existingIDs, ids...)
+	} else {
+ d.blobsByHeight[height] = ids
+	}
 	d.timestampsByHeight[height] = time.Now()
 
 	return ids, nil

core/da/dummy_test.go

@@ -3,11 +3,16 @@ package da
 import (
 	"context"
 	"testing"
+	"time"
 )
 
 func TestDummyDA(t *testing.T) {
+	testDABlockTime := 100 * time.Millisecond
 	// Create a new DummyDA instance with a max blob size of 1024 bytes
-	dummyDA := NewDummyDA(1024, 0, 0)
+	dummyDA := NewDummyDA(1024, 0, 0, testDABlockTime)
+	dummyDA.StartHeightTicker()
+	defer dummyDA.StopHeightTicker()
+	// Height is always 0
 	ctx := context.Background()
 
 	// Test MaxBlobSize
@@ -25,6 +30,7 @@ func TestDummyDA(t *testing.T) {
  []byte("test blob 2"),
 	}
 	ids, err := dummyDA.Submit(ctx, blobs, 0, nil)
+	time.Sleep(testDABlockTime)
 	if err != nil {
  t.Fatalf("Submit failed: %v", err)
 	}
@@ -47,7 +53,7 @@ func TestDummyDA(t *testing.T) {
 	}
 
 	// Test GetIDs
-	result, err := dummyDA.GetIDs(ctx, 0, nil)
+	result, err := dummyDA.GetIDs(ctx, 1, nil)
 	if err != nil {
  t.Fatalf("GetIDs failed: %v", err)
 	}

core/da/errors.go

@@ -11,5 +11,5 @@ var (
 	ErrTxAlreadyInMempool = errors.New("tx already in mempool")
 	ErrTxIncorrectAccountSequence = errors.New("incorrect account sequence")
 	ErrContextDeadline = errors.New("context deadline")
-ErrFutureHeight= errors.New("future height")
+ErrHeightFromFuture = errors.New("given height is from the future")
 )

core/sequencer/dummy.go

@@ -15,14 +15,14 @@ var _ Sequencer = (*DummySequencer)(nil)
 // DummySequencer is a dummy implementation of the Sequencer interface for testing
 type DummySequencer struct {
 	mu sync.RWMutex
-	batches map[string]*Batch
+	batches map[string][]*Batch
 	batchSubmissionChan chan Batch
 }
 
 // NewDummySequencer creates a new dummy Sequencer instance
 func NewDummySequencer() *DummySequencer {
 	return &DummySequencer{
- batches: make(map[string]*Batch),
+ batches: make(map[string][]*Batch),
 	}
 }
 
@@ -31,21 +31,29 @@ func (s *DummySequencer) SubmitBatchTxs(ctx context.Context, req SubmitBatchTxsR
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	s.batches[string(req.Id)] = req.Batch
-	if req.Batch != nil && len(req.Batch.Transactions) > 0 {
- s.batchSubmissionChan <- *req.Batch
+	if req.Batch != nil {
+ rollupKey := string(req.Id)
+ s.batches[rollupKey] = append(s.batches[rollupKey], req.Batch)
+ if len(req.Batch.Transactions) > 0 {
+ s.batchSubmissionChan <- *req.Batch
+ }
 	}
 	return &SubmitBatchTxsResponse{}, nil
 }
 
 // GetNextBatch gets the next batch from the sequencer
 func (s *DummySequencer) GetNextBatch(ctx context.Context, req GetNextBatchRequest) (*GetNextBatchResponse, error) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
+	s.mu.Lock()
+	defer s.mu.Unlock()
 
-	batch, ok := s.batches[string(req.Id)]
-	if !ok {
+	rollupKey := string(req.Id)
+	batches := s.batches[rollupKey]
+	var batch *Batch
+	if len(batches) == 0 {
  batch = &Batch{Transactions: nil}
+	} else {
+ batch = batches[0]
+ s.batches[rollupKey] = batches[1:]
 	}
 
 	return &GetNextBatchResponse{

core/sequencer/dummy_test.go

@@ -302,18 +302,27 @@ func TestDummySequencer_BatchOverwrite(t *testing.T) {
  t.Fatalf("SubmitBatchTxs should not return an error: %v", err)
 	}
 
-	// Get the batch and verify it's the second one
+	// Get the first batch and verify it's batch1
 	getResp, err := seq.GetNextBatch(ctx, GetNextBatchRequest{
  Id: ID,
 	})
 
 	if err != nil {
  t.Fatalf("GetNextBatch should not return an error: %v", err)
 	}
-	if !reflect.DeepEqual(batch2, getResp.Batch) {
- t.Fatal("Retrieved batch should be the most recently submitted one")
+	if !reflect.DeepEqual(batch1, getResp.Batch) {
+ t.Fatal("Retrieved batch should be the first submitted one (batch1)")
 	}
-	if reflect.DeepEqual(batch1, getResp.Batch) {
- t.Fatal("Retrieved batch should not be the first submitted one")
+
+	// Get the second batch and verify it's batch2
+	getResp, err = seq.GetNextBatch(ctx, GetNextBatchRequest{
+ Id: ID,
+	})
+
+	if err != nil {
+ t.Fatalf("GetNextBatch should not return an error: %v", err)
+	}
+	if !reflect.DeepEqual(batch2, getResp.Batch) {
+ t.Fatal("Retrieved batch should be the second submitted one (batch2)")
 	}
 }

da/cmd/local-da/local.go

@@ -112,7 +112,7 @@ func (d *LocalDA) GetIDs(ctx context.Context, height uint64, _ []byte) (*coreda.
 
 	if height > d.height {
  d.logger.Error("GetIDs: height in future", "requested", height, "current", d.height)
- return nil, fmt.Errorf("height %d is in the future: %w", height, coreda.ErrFutureHeight)
+ return nil, fmt.Errorf("height %d is in the future: %w", height, coreda.ErrHeightFromFuture)
 	}
 
 	kvps, ok := d.data[height]