Add TestTxGoMultipleDAIncluded

Author: Manav-Aggarwal

core/da/dummy.go

@@ -162,7 +162,7 @@ func (d *DummyDA) SubmitWithOptions(ctx context.Context, blobs []Blob, gasPrice
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
-	height := d.currentHeight
+	height := d.currentHeight + 1
 	ids := make([]ID, 0, len(blobs))
 	var currentSize uint64
 
@@ -201,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

@@ -7,8 +7,11 @@ import (
 )
 
 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, 10*time.Second)
+	dummyDA := NewDummyDA(1024, 0, 0, testDABlockTime)
+	dummyDA.StartHeightTicker()
+	defer dummyDA.StopHeightTicker()
 	// Height is always 0
 	ctx := context.Background()
 
@@ -27,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)
 	}
@@ -49,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/sequencer/dummy_test.go

@@ -316,7 +316,7 @@ func TestDummySequencer_BatchOverwrite(t *testing.T) {
 
 	// Get the second batch and verify it's batch2
 	getResp, err = seq.GetNextBatch(ctx, GetNextBatchRequest{
-		RollupId: rollupID,
+		Id: ID,
 	})
 
 	if err != nil {

da/jsonrpc/client.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
-	"strings"
 
 	"cosmossdk.io/log"
 	"github.com/filecoin-project/go-jsonrpc"
@@ -61,7 +60,7 @@ func (api *API) GetIDs(ctx context.Context, height uint64, _ []byte) (*da.GetIDs
 			api.Logger.Debug("RPC call indicates blobs not found", "method", "GetIDs", "height", height)
 			return nil, err // Return the specific ErrBlobNotFound
 		}
-		if strings.Contains(err.Error(), da.ErrHeightFromFuture.Error()) {
+		if errors.Is(err, da.ErrHeightFromFuture) {
 			api.Logger.Debug("RPC call indicates height from future", "method", "GetIDs", "height", height)
 			return nil, err // Return the specific ErrHeightFromFuture
 		}

da/jsonrpc/proxy_test.go

@@ -73,6 +73,7 @@ func TestProxy(t *testing.T) {
 	t.Run("Given height is from the future", func(t *testing.T) {
 		HeightFromFutureTest(t, &client.DA)
 	})
+	dummy.StopHeightTicker()
 }
 
 // BasicDATest tests round trip of messages to DA and back.
@@ -137,6 +138,7 @@ func GetIDsTest(t *testing.T, d coreda.DA) {
 
 	ctx := t.Context()
 	ids, err := d.Submit(ctx, msgs, 0, testNamespace)
+	time.Sleep(getTestDABlockTime())
 	assert.NoError(t, err)
 	assert.Len(t, ids, len(msgs))
 	found := false
@@ -145,7 +147,7 @@ func GetIDsTest(t *testing.T, d coreda.DA) {
 	// To Keep It Simple: we assume working with DA used exclusively for this test (mock, devnet, etc)
 	// As we're the only user, we don't need to handle external data (that could be submitted in real world).
 	// There is no notion of height, so we need to scan the DA to get test data back.
-	for i := uint64(0); !found && !time.Now().After(end); i++ {
+	for i := uint64(1); !found && !time.Now().After(end); i++ {
 		ret, err := d.GetIDs(ctx, i, []byte{})
 		if err != nil {
 			if strings.Contains(err.Error(), coreda.ErrHeightFromFuture.Error()) {
@@ -161,11 +163,19 @@ func GetIDsTest(t *testing.T, d coreda.DA) {
 
 			// Submit ensures atomicity of batch, so it makes sense to compare actual blobs (bodies) only when lengths
 			// of slices is the same.
-			if len(blobs) == len(msgs) {
+			if len(blobs) >= len(msgs) {
 				found = true
-				for b := 0; b < len(blobs); b++ {
-					if !bytes.Equal(blobs[b], msgs[b]) {
+				for _, msg := range msgs {
+					msgFound := false
+					for _, blob := range blobs {
+						if bytes.Equal(blob, msg) {
+							msgFound = true
+							break
+						}
+					}
+					if !msgFound {
 						found = false
+						break
 					}
 				}
 			}

node/full_node_integration_test.go

@@ -11,11 +11,61 @@ import (
 	rollkitconfig "github.com/rollkit/rollkit/pkg/config"
 )
 
-// TestTxGossipingAndAggregation tests that transactions are gossiped and blocks are aggregated and synced across multiple nodes.
+// TestTxGossipingMultipleNodesNoDA tests that transactions are gossiped and blocks are sequenced and synced across multiple nodes without the DA layer over P2P.
+// It creates 4 nodes (1 sequencer, 3 full nodes), injects a transaction, waits for all nodes to sync, and asserts block equality.
+func TestTxGossipingMultipleNodesNoDA(t *testing.T) {
+	require := require.New(t)
+	config := getTestConfig(t, 1)
+	// Set the DA block time to a very large value to ensure that the DA layer is not used
+	config.DA.BlockTime = rollkitconfig.DurationWrapper{Duration: 100 * time.Second}
+	numNodes := 4
+	nodes, cleanups := createNodesWithCleanup(t, numNodes, config)
+	for _, cleanup := range cleanups {
+		defer cleanup()
+	}
+
+	ctxs, cancels := createNodeContexts(numNodes)
+	var runningWg sync.WaitGroup
+
+	// Start only the sequencer first
+	startNodeInBackground(t, nodes, ctxs, &runningWg, 0)
+
+	// Wait for the first block to be produced by the sequencer
+	err := waitForFirstBlock(nodes[0], Header)
+	require.NoError(err)
+
+	// Verify block manager is properly initialized
+	require.NotNil(nodes[0].blockManager, "Block manager should be initialized")
+
+	// Start the other nodes
+	for i := 1; i < numNodes; i++ {
+		startNodeInBackground(t, nodes, ctxs, &runningWg, i)
+	}
+
+	// Inject a transaction into the sequencer's executor
+	executor := nodes[0].blockManager.GetExecutor().(*coreexecutor.DummyExecutor)
+	executor.InjectTx([]byte("test tx"))
+
+	blocksToWaitFor := uint64(5)
+	// Wait for all nodes to reach at least 5 blocks with DA inclusion
+	for _, node := range nodes {
+		require.NoError(waitForAtLeastNBlocks(node, blocksToWaitFor, Store))
+	}
+
+	// Shutdown all nodes and wait
+	shutdownAndWait(t, cancels, &runningWg, 5*time.Second)
+
+	// Assert that all nodes have the same block up to height blocksToWaitFor
+	assertAllNodesSynced(t, nodes, blocksToWaitFor)
+}
+
+// TestTxGossipingMultipleNodesDAIncluded tests that transactions are gossiped and blocks are sequenced and synced across multiple nodes only using DA. P2P gossiping is disabled.
 // It creates 4 nodes (1 sequencer, 3 full nodes), injects a transaction, waits for all nodes to sync with DA inclusion, and asserts block equality.
-func TestTxGossipingAndAggregation(t *testing.T) {
+func TestTxGossipingMultipleNodesDAIncluded(t *testing.T) {
 	require := require.New(t)
 	config := getTestConfig(t, 1)
+	// Disable P2P gossiping
+	config.P2P.Peers = "none"
 
 	numNodes := 4
 	nodes, cleanups := createNodesWithCleanup(t, numNodes, config)
@@ -68,8 +118,10 @@ func TestFastDASync(t *testing.T) {
 
 	// Set up two nodes with different block and DA block times
 	config := getTestConfig(t, 1)
-	config.Node.BlockTime = rollkitconfig.DurationWrapper{Duration: 200 * time.Millisecond}
-	config.DA.BlockTime = rollkitconfig.DurationWrapper{Duration: 50 * time.Millisecond}
+	// Set the block time to 2 seconds and the DA block time to 1 second
+	// Note: these are large values to avoid test failures due to slow CI machines
+	config.Node.BlockTime = rollkitconfig.DurationWrapper{Duration: 2 * time.Second}
+	config.DA.BlockTime = rollkitconfig.DurationWrapper{Duration: 1 * time.Second}
 
 	nodes, cleanups := createNodesWithCleanup(t, 2, config)
 	for _, cleanup := range cleanups {
@@ -83,7 +135,7 @@ func TestFastDASync(t *testing.T) {
 	startNodeInBackground(t, nodes, ctxs, &runningWg, 0)
 
 	// Wait for the first node to produce a few blocks
-	blocksToWaitFor := uint64(5)
+	blocksToWaitFor := uint64(2)
 	require.NoError(waitForAtLeastNDAIncludedHeight(nodes[0], blocksToWaitFor))
 
 	// Now start the second node and time its sync
@@ -94,7 +146,7 @@ func TestFastDASync(t *testing.T) {
 	syncDuration := time.Since(start)
 
 	// Ensure node syncs within a small delta of DA block time
-	delta := 75 * time.Millisecond
+	delta := 250 * time.Millisecond
 	require.Less(syncDuration, config.DA.BlockTime.Duration+delta, "Block sync should be faster than DA block time")
 
 	// Verify both nodes are synced and that the synced block is DA-included
@@ -276,7 +328,7 @@ func testSingleSequencerTwoFullNodes(t *testing.T, source Source) {
 	shutdownAndWait(t, cancels, &runningWg, 5*time.Second)
 }
 
-// testSingleSequencerSingleFullNode sets up a single sequencer and a single full node with a trusted hash, starts the sequencer, waits for it to produce a block, then starts the full node with the trusted hash.
+// testSingleSequencerSingleFullNodeTrustedHash sets up a single sequencer and a single full node with a trusted hash, starts the sequencer, waits for it to produce a block, then starts the full node with the trusted hash.
 // It waits for both nodes to reach a target block height (using the provided 'source' to determine block inclusion), verifies that both nodes are fully synced, and then shuts them down.
 func testSingleSequencerSingleFullNodeTrustedHash(t *testing.T, source Source) {
 	require := require.New(t)

node/helpers_test.go

@@ -42,12 +42,16 @@ const (
 )
 
 // createTestComponents creates test components for node initialization
-func createTestComponents(t *testing.T, config rollkitconfig.Config) (coreexecutor.Executor, coresequencer.Sequencer, coreda.DA, *p2p.Client, datastore.Batching, *key.NodeKey) {
+func createTestComponents(t *testing.T, config rollkitconfig.Config) (coreexecutor.Executor, coresequencer.Sequencer, coreda.DA, *p2p.Client, datastore.Batching, *key.NodeKey, func()) {
 	executor := coreexecutor.NewDummyExecutor()
 	sequencer := coresequencer.NewDummySequencer()
 	dummyDA := coreda.NewDummyDA(100_000, 0, 0, config.DA.BlockTime.Duration)
 	dummyDA.StartHeightTicker()
 
+	stopDAHeightTicker := func() {
+		dummyDA.StopHeightTicker()
+	}
+
 	// Create genesis and keys for P2P client
 	_, genesisValidatorKey, _ := types.GetGenesisWithPrivkey("test-chain")
 	p2pKey := &key.NodeKey{
@@ -59,7 +63,7 @@ func createTestComponents(t *testing.T, config rollkitconfig.Config) (coreexecut
 	require.NotNil(t, p2pClient)
 	ds := dssync.MutexWrap(datastore.NewMapDatastore())
 
-	return executor, sequencer, dummyDA, p2pClient, ds, p2pKey
+	return executor, sequencer, dummyDA, p2pClient, ds, p2pKey, stopDAHeightTicker
 }
 
 func getTestConfig(t *testing.T, n int) rollkitconfig.Config {
@@ -97,8 +101,7 @@ func createNodeWithCleanup(t *testing.T, config rollkitconfig.Config) (*FullNode
 	remoteSigner, err := remote_signer.NewNoopSigner(genesisValidatorKey)
 	require.NoError(t, err)
 
-	executor, sequencer, dac, p2pClient, ds, _ := createTestComponents(t, config)
-
+	executor, sequencer, dac, p2pClient, ds, _, stopDAHeightTicker := createTestComponents(t, config)
 	require.NoError(t, err)
 
 	node, err := NewNode(
@@ -120,6 +123,7 @@ func createNodeWithCleanup(t *testing.T, config rollkitconfig.Config) (*FullNode
 	cleanup := func() {
 		// Cancel the context to stop the node
 		cancel()
+		stopDAHeightTicker()
 	}
 
 	return node.(*FullNode), cleanup
@@ -141,7 +145,8 @@ func createNodesWithCleanup(t *testing.T, num int, config rollkitconfig.Config)
 	require.NoError(err)
 
 	aggListenAddress := config.P2P.ListenAddress
-	executor, sequencer, dac, p2pClient, ds, aggP2PKey := createTestComponents(t, config)
+	aggPeers := config.P2P.Peers
+	executor, sequencer, dac, p2pClient, ds, aggP2PKey, stopDAHeightTicker := createTestComponents(t, config)
 	aggPeerID, err := peer.IDFromPrivateKey(aggP2PKey.PrivKey)
 	require.NoError(err)
 
@@ -164,18 +169,24 @@ func createNodesWithCleanup(t *testing.T, num int, config rollkitconfig.Config)
 	cleanup := func() {
 		// Cancel the context to stop the node
 		aggCancel()
+		stopDAHeightTicker()
 	}
 
 	nodes[0], cleanups[0] = aggNode.(*FullNode), cleanup
 	config.Node.Aggregator = false
-	aggPeerAddress := fmt.Sprintf("%s/p2p/%s", aggListenAddress, aggPeerID.Loggable()["peerID"].(string))
-	peersList := []string{aggPeerAddress}
+	peersList := []string{}
+	if aggPeers != "none" {
+		aggPeerAddress := fmt.Sprintf("%s/p2p/%s", aggListenAddress, aggPeerID.Loggable()["peerID"].(string))
+		peersList = append(peersList, aggPeerAddress)
+	}
 	for i := 1; i < num; i++ {
 		ctx, cancel := context.WithCancel(context.Background())
-		config.P2P.Peers = strings.Join(peersList, ",")
+		if aggPeers != "none" {
+			config.P2P.Peers = strings.Join(peersList, ",")
+		}
 		config.P2P.ListenAddress = fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", 40001+i)
 		config.RPC.Address = fmt.Sprintf("127.0.0.1:%d", 8001+i)
-		executor, sequencer, _, p2pClient, _, nodeP2PKey := createTestComponents(t, config)
+		executor, sequencer, _, p2pClient, _, nodeP2PKey, stopDAHeightTicker := createTestComponents(t, config)
 		node, err := NewNode(
 			ctx,
 			config,
@@ -194,6 +205,7 @@ func createNodesWithCleanup(t *testing.T, num int, config rollkitconfig.Config)
 		cleanup := func() {
 			// Cancel the context to stop the node
 			cancel()
+			stopDAHeightTicker()
 		}
 		nodes[i], cleanups[i] = node.(*FullNode), cleanup
 		nodePeerID, err := peer.IDFromPrivateKey(nodeP2PKey.PrivKey)

pkg/cmd/run_node_test.go

@@ -24,11 +24,14 @@ import (
 
 const MockDANamespace = "test"
 
-func createTestComponents(_ context.Context, t *testing.T) (coreexecutor.Executor, coresequencer.Sequencer, coreda.DA, signer.Signer, *p2p.Client, datastore.Batching) {
+func createTestComponents(_ context.Context, t *testing.T) (coreexecutor.Executor, coresequencer.Sequencer, coreda.DA, signer.Signer, *p2p.Client, datastore.Batching, func()) {
 	executor := coreexecutor.NewDummyExecutor()
 	sequencer := coresequencer.NewDummySequencer()
 	dummyDA := coreda.NewDummyDA(100_000, 0, 0, 10*time.Second)
 	dummyDA.StartHeightTicker()
+	stopDAHeightTicker := func() {
+		dummyDA.StopHeightTicker()
+	}
 	tmpDir := t.TempDir()
 	keyProvider, err := filesigner.CreateFileSystemSigner(filepath.Join(tmpDir, "config"), []byte{})
 	if err != nil {
@@ -38,7 +41,7 @@ func createTestComponents(_ context.Context, t *testing.T) (coreexecutor.Executo
 	p2pClient := &p2p.Client{}
 	ds := datastore.NewMapDatastore()
 
-	return executor, sequencer, dummyDA, keyProvider, p2pClient, ds
+	return executor, sequencer, dummyDA, keyProvider, p2pClient, ds, stopDAHeightTicker
 }
 
 func TestParseFlags(t *testing.T) {
@@ -78,7 +81,8 @@ func TestParseFlags(t *testing.T) {
 
 	args := append([]string{"start"}, flags...)
 
-	executor, sequencer, dac, keyProvider, p2pClient, ds := createTestComponents(context.Background(), t)
+	executor, sequencer, dac, keyProvider, p2pClient, ds, stopDAHeightTicker := createTestComponents(context.Background(), t)
+	defer stopDAHeightTicker()
 
 	nodeConfig := rollconf.DefaultConfig
 	nodeConfig.RootDir = t.TempDir()
@@ -154,7 +158,8 @@ func TestAggregatorFlagInvariants(t *testing.T) {
 	for i, flags := range flagVariants {
 		args := append([]string{"start"}, flags...)
 
-		executor, sequencer, dac, keyProvider, p2pClient, ds := createTestComponents(context.Background(), t)
+		executor, sequencer, dac, keyProvider, p2pClient, ds, stopDAHeightTicker := createTestComponents(context.Background(), t)
+		defer stopDAHeightTicker()
 
 		nodeConfig := rollconf.DefaultConfig
 		nodeConfig.RootDir = t.TempDir()
@@ -189,7 +194,8 @@ func TestDefaultAggregatorValue(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			executor, sequencer, dac, keyProvider, p2pClient, ds := createTestComponents(context.Background(), t)
+			executor, sequencer, dac, keyProvider, p2pClient, ds, stopDAHeightTicker := createTestComponents(context.Background(), t)
+			defer stopDAHeightTicker()
 
 			nodeConfig := rollconf.DefaultConfig
 
@@ -268,7 +274,8 @@ func TestCentralizedAddresses(t *testing.T) {
 		"--rollkit.da.address=http://central-da:26657",
 	}
 
-	executor, sequencer, dac, keyProvider, p2pClient, ds := createTestComponents(context.Background(), t)
+	executor, sequencer, dac, keyProvider, p2pClient, ds, stopDAHeightTicker := createTestComponents(context.Background(), t)
+	defer stopDAHeightTicker()
 
 	cmd := newRunNodeCmd(t.Context(), executor, sequencer, dac, keyProvider, p2pClient, ds, nodeConfig)
 	_ = cmd.Flags().Set(rollconf.FlagRootDir, "custom/root/dir")
@@ -291,7 +298,9 @@ func TestStartNodeErrors(t *testing.T) {
 	logger := log.NewNopLogger() // Use NopLogger for tests unless specific logging output is needed
 
 	// Common setup
-	executor, sequencer, dac, _, p2pClient, ds := createTestComponents(baseCtx, t)
+	executor, sequencer, dac, _, p2pClient, ds, stopDAHeightTicker := createTestComponents(baseCtx, t)
+	defer stopDAHeightTicker()
+
 	tmpDir := t.TempDir()
 
 	// Create a dummy genesis file for successful load cases