test: Add back missing node integration tests (#2280)

<!--
Please read and fill out this form before submitting your PR.

Please make sure you have reviewed our contributors guide before
submitting your
first PR.

NOTE: PR titles should follow semantic commits:
https://www.conventionalcommits.org/en/v1.0.0/
-->

## Overview

This PR adds/updates the following integration tests for full node and
sequencer behavior in Rollkit:

**TestTxGossipingMultipleNodesNoDA**
Validates transaction gossiping and block synchronization across
multiple nodes (1 sequencer, 3 full nodes) over P2P, with the DA layer
disabled.

**TestTxGossipingMultipleNodesDAIncluded**
Ensures transaction gossiping and block synchronization across multiple
nodes (1 sequencer, 3 full nodes) using only the DA layer, with P2P
gossiping disabled.

**TestFastDASync**
Verifies that a new node can quickly synchronize with the DA layer using
fast sync, ensuring the sync speed is within a small delta of the DA
block time.

**TestSingleSequencerTwoFullNodesBlockSyncSpeed**
Ensures that block synchronization among nodes is not bottlenecked by DA
block time, using a fast block time and a slow DA block time.

**TestDataExchange**
Verifies data exchange and synchronization between nodes in various
network topologies, including single and multiple full nodes, and
trusted hash scenarios.

**TestHeaderExchange**
Verifies header exchange and synchronization between nodes in various
network topologies, including single and multiple full nodes, and
trusted hash scenarios.

**TestTwoChainsInOneNamespace**
Verifies that two chains in the same namespace (with same or different
chain IDs) can coexist and synchronize independently without issues.

Also, fixes **TestMaxPendingHeaders** behavior.

<!-- 
Please provide an explanation of the PR, including the appropriate
context,
background, goal, and rationale. If there is an issue with this
information,
please provide a tl;dr and link the issue. 

Ex: Closes #<issue number>
-->


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Added height simulation and multi-batch handling in dummy DA and
sequencer components for more realistic testing.
- Introduced multi-node orchestration helpers to facilitate complex
integration tests and synchronization verification.
- Expanded full node integration tests to verify transaction
propagation, block production, and DA inclusion across various
scenarios.
- Enhanced node shutdown logging for better observability during tests.

- **Bug Fixes**
- Corrected data inclusion checks to account for current sync height,
preventing false positives.
- Improved error handling for future height queries in DA clients and
JSON-RPC, providing clearer error messages.

- **Documentation**
- Updated comments and terminology to reflect new data sync naming
conventions and reference cleanup.

- **Tests**
- Extended testing framework with multi-node setup, synchronization, and
timing controls.
- Added tests for fast sync, block speed, and multiple chain
coexistence.
- Refined test timing, context management, and error handling for
robustness.

- **Chores**
- Updated dependency versions, especially core module, to support new
features.
- Removed obsolete test files and simplified test helper functions for
maintainability.

- **Style**
- Improved debug logs for node shutdown and data retrieval, aiding
troubleshooting and observability.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: Manav-Aggarwal

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")
 )