only remove successfully submitted blocks from the pending blocks instead of reset (#1376) (#1394)

cherry-picking

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

- **New Features**
- Enhanced block submission process with updated logic for handling and
caching submitted blocks.
- Improved block management with a new method for removing specific
submitted blocks.

- **Refactor**
- Transitioned `PendingBlocks` storage from a slice to a map for
optimized block handling.
- Streamlined the retrieval of pending blocks with a new sorting
mechanism based on block height.

- **Tests**
- Added comprehensive test cases for `PendingBlocks` to ensure correct
functionality when adding and removing blocks.
- Introduced a new integration test to verify the successful submission
of blocks in a simulated environment.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Ganesha Upadhyaya <gupadhyaya@Ganeshas-MacBook-Pro-2.local>

Author: gupadhyaya

block/manager.go

@@ -746,7 +746,11 @@ func (m *Manager) submitBlocksToDA(ctx context.Context) error {
 			if int(res.SubmittedCount) == len(blocks) {
 				submitted = true
 			}
-			m.pendingBlocks.resetPendingBlocks(res.SubmittedCount)
+			submittedBlocks := blocks[:res.SubmittedCount]
+			for _, block := range submittedBlocks {
+				m.blockCache.setDAIncluded(block.Hash().String())
+			}
+			m.pendingBlocks.removeSubmittedBlocks(submittedBlocks)
 		case da.StatusError, da.StatusNotFound, da.StatusUnknown:
 			m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
 			time.Sleep(backoff)

block/pending_blocks.go

@@ -1,47 +1,57 @@
 package block
 
 import (
+	"sort"
 	"sync"
 
 	"github.com/rollkit/rollkit/types"
 )
 
 // PendingBlocks maintains blocks that need to be published to DA layer
 type PendingBlocks struct {
-	pendingBlocks []*types.Block
+	pendingBlocks map[uint64]*types.Block
 	mtx           *sync.RWMutex
 }
 
 // NewPendingBlocks returns a new PendingBlocks struct
 func NewPendingBlocks() *PendingBlocks {
 	return &PendingBlocks{
-		pendingBlocks: make([]*types.Block, 0),
+		pendingBlocks: make(map[uint64]*types.Block),
 		mtx:           new(sync.RWMutex),
 	}
 }
 
+// getPendingBlocks returns a sorted slice of pending blocks
+// that need to be published to DA layer in order of block height
 func (pb *PendingBlocks) getPendingBlocks() []*types.Block {
 	pb.mtx.RLock()
 	defer pb.mtx.RUnlock()
-	return pb.pendingBlocks
+	blocks := make([]*types.Block, 0, len(pb.pendingBlocks))
+	for _, block := range pb.pendingBlocks {
+		blocks = append(blocks, block)
+	}
+	sort.Slice(blocks, func(i, j int) bool {
+		return blocks[i].Height() < blocks[j].Height()
+	})
+	return blocks
 }
 
 func (pb *PendingBlocks) isEmpty() bool {
-	pendingBlocks := pb.getPendingBlocks()
-	return len(pendingBlocks) == 0
+	pb.mtx.RLock()
+	defer pb.mtx.RUnlock()
+	return len(pb.pendingBlocks) == 0
 }
 
 func (pb *PendingBlocks) addPendingBlock(block *types.Block) {
 	pb.mtx.Lock()
 	defer pb.mtx.Unlock()
-	pb.pendingBlocks = append(pb.pendingBlocks, block)
+	pb.pendingBlocks[block.Height()] = block
 }
 
-func (pb *PendingBlocks) resetPendingBlocks(submitted uint64) {
+func (pb *PendingBlocks) removeSubmittedBlocks(blocks []*types.Block) {
 	pb.mtx.Lock()
 	defer pb.mtx.Unlock()
-	if submitted > uint64(len(pb.pendingBlocks)) {
-		submitted = uint64(len(pb.pendingBlocks))
+	for _, block := range blocks {
+		delete(pb.pendingBlocks, block.Height())
 	}
-	pb.pendingBlocks = pb.pendingBlocks[submitted:]
 }

block/pending_blocks_test.go

@@ -0,0 +1,74 @@
+package block
+
+import (
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/rollkit/rollkit/types"
+)
+
+func TestGetPendingBlocks(t *testing.T) {
+	require := require.New(t)
+	pb := NewPendingBlocks()
+	for i := uint64(0); i < 5; i++ {
+		pb.addPendingBlock(types.GetRandomBlock(i, 0))
+	}
+	blocks := pb.getPendingBlocks()
+	require.True(sort.SliceIsSorted(blocks, func(i, j int) bool {
+		return blocks[i].Height() < blocks[j].Height()
+	}))
+}
+
+func TestRemoveSubmittedBlocks(t *testing.T) {
+	require := require.New(t)
+	pb := NewPendingBlocks()
+	for i := uint64(0); i < 5; i++ {
+		pb.addPendingBlock(types.GetRandomBlock(i, 0))
+	}
+	blocks := pb.getPendingBlocks()
+	pb.removeSubmittedBlocks(blocks)
+	require.True(pb.isEmpty())
+}
+
+func TestRemoveSubsetOfBlocks(t *testing.T) {
+	require := require.New(t)
+	pb := NewPendingBlocks()
+	for i := uint64(0); i < 5; i++ {
+		pb.addPendingBlock(types.GetRandomBlock(i, 0))
+	}
+	// Remove blocks with height 1 and 2
+	pb.removeSubmittedBlocks([]*types.Block{
+		types.GetRandomBlock(1, 0),
+		types.GetRandomBlock(2, 0),
+	})
+	remainingBlocks := pb.getPendingBlocks()
+	require.Len(remainingBlocks, 3, "There should be 3 blocks remaining")
+	for _, block := range remainingBlocks {
+		require.Contains([]uint64{0, 3, 4}, block.Height(), "Only blocks with height 0, 3, and 4 should remain")
+	}
+}
+
+func TestRemoveAllBlocksAndVerifyEmpty(t *testing.T) {
+	require := require.New(t)
+	pb := NewPendingBlocks()
+	for i := uint64(0); i < 5; i++ {
+		pb.addPendingBlock(types.GetRandomBlock(i, 0))
+	}
+	// Remove all blocks
+	pb.removeSubmittedBlocks(pb.getPendingBlocks())
+	require.True(pb.isEmpty(), "PendingBlocks should be empty after removing all blocks")
+}
+
+func TestRemoveBlocksFromEmptyPendingBlocks(t *testing.T) {
+	require := require.New(t)
+	pb := NewPendingBlocks()
+	// Attempt to remove blocks from an empty PendingBlocks
+	require.NotPanics(func() {
+		pb.removeSubmittedBlocks([]*types.Block{
+			types.GetRandomBlock(1, 0),
+			types.GetRandomBlock(2, 0),
+		})
+	}, "Removing blocks from an empty PendingBlocks should not cause a panic")
+}

node/full_node_integration_test.go

@@ -370,6 +370,41 @@ func TestHeaderExchange(t *testing.T) {
 	t.Run("SingleAggregatorSingleFullNodeSingleLightNode", testSingleAggregatorSingleFullNodeSingleLightNode)
 }
 
+func TestSubmitBlocksToDA(t *testing.T) {
+	require := require.New(t)
+
+	clientNodes := 1
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	nodes, _ := createNodes(
+		ctx,
+		context.Background(),
+		clientNodes,
+		config.BlockManagerConfig{
+			DABlockTime: 20 * time.Millisecond,
+			BlockTime:   10 * time.Millisecond,
+		},
+		t,
+	)
+	seq := nodes[0]
+	require.NoError(seq.Start())
+	defer func() {
+		require.NoError(seq.Stop())
+	}()
+
+	timer := time.NewTimer(5 * seq.nodeConfig.DABlockTime)
+	<-timer.C
+
+	numberOfBlocksToSyncTill := seq.Store.Height()
+
+	//Make sure all produced blocks made it to DA
+	for i := uint64(1); i <= numberOfBlocksToSyncTill; i++ {
+		block, err := seq.Store.GetBlock(i)
+		require.NoError(err)
+		require.True(seq.blockManager.IsDAIncluded(block.Hash()), block.Height())
+	}
+}
+
 func testSingleAggregatorSingleFullNode(t *testing.T, source Source) {
 	require := require.New(t)