minor fixes

Author: gupadhyaya

block/batch.go

@@ -4,27 +4,27 @@ import (
 	"sync"
 )
 
-// BatchQueue is a queue of transactions
+// BatchQueue is a queue of transaction batches with timestamps
 type BatchQueue struct {
 	queue []BatchWithTime
 	mu    sync.Mutex
 }
 
-// NewBatchQueue creates a new TransactionQueue
+// NewBatchQueue creates a new BatchQueue
 func NewBatchQueue() *BatchQueue {
 	return &BatchQueue{
 		queue: make([]BatchWithTime, 0),
 	}
 }
 
-// AddBatch adds a new transaction to the queue
+// AddBatch adds a new batch to the queue
 func (bq *BatchQueue) AddBatch(batch BatchWithTime) {
 	bq.mu.Lock()
 	defer bq.mu.Unlock()
 	bq.queue = append(bq.queue, batch)
 }
 
-// Next returns the next transaction in the queue
+// Next returns the next batch in the queue
 func (bq *BatchQueue) Next() *BatchWithTime {
 	bq.mu.Lock()
 	defer bq.mu.Unlock()

block/manager.go

@@ -28,7 +28,7 @@ import (
 	goheaderstore "github.com/celestiaorg/go-header/store"
 
 	"github.com/rollkit/go-sequencing"
-	seqGRPC "github.com/rollkit/go-sequencing/proxy/grpc"
+	"github.com/rollkit/go-sequencing/proxy/grpc"
 	"github.com/rollkit/rollkit/config"
 	"github.com/rollkit/rollkit/da"
 	"github.com/rollkit/rollkit/mempool"
@@ -161,7 +161,7 @@ type Manager struct {
 	// in the DA
 	daIncludedHeight atomic.Uint64
 	// grpc client for sequencing middleware
-	seqClient     *seqGRPC.Client
+	seqClient     *grpc.Client
 	lastBatchHash []byte
 	bq            *BatchQueue
 }
@@ -199,7 +199,7 @@ func NewManager(
 	store store.Store,
 	mempool mempool.Mempool,
 	mempoolReaper *mempool.CListMempoolReaper,
-	seqClient *seqGRPC.Client,
+	seqClient *grpc.Client,
 	proxyApp proxy.AppConnConsensus,
 	dalc *da.DAClient,
 	eventBus *cmtypes.EventBus,
@@ -815,9 +815,6 @@ func (m *Manager) getHeadersFromHeaderStore(ctx context.Context, startHeight, en
 	if startHeight > endHeight {
 		return nil, fmt.Errorf("startHeight (%d) is greater than endHeight (%d)", startHeight, endHeight)
 	}
-	if startHeight == 0 {
-		startHeight++
-	}
 	headers := make([]*types.SignedHeader, endHeight-startHeight+1)
 	for i := startHeight; i <= endHeight; i++ {
 		header, err := m.headerStore.GetByHeight(ctx, i)
@@ -833,9 +830,6 @@ func (m *Manager) getDataFromDataStore(ctx context.Context, startHeight, endHeig
 	if startHeight > endHeight {
 		return nil, fmt.Errorf("startHeight (%d) is greater than endHeight (%d)", startHeight, endHeight)
 	}
-	if startHeight == 0 {
-		startHeight++
-	}
 	data := make([]*types.Data, endHeight-startHeight+1)
 	for i := startHeight; i <= endHeight; i++ {
 		d, err := m.dataStore.GetByHeight(ctx, i)

block/pending_test.go

@@ -34,7 +34,7 @@ func TestPendingBlocks(t *testing.T) {
 		},
 		{
 			name: "mock successful DA submission of some blocks by manually setting last submitted height",
-			init: fillWithBlocks,
+			init: fillWithBlockData,
 			exec: func(ctx context.Context, t *testing.T, pb *PendingHeaders) {
 				pb.lastSubmittedHeight.Store(testHeight)
 			},
@@ -43,7 +43,7 @@ func TestPendingBlocks(t *testing.T) {
 		},
 		{
 			name: "mock successful DA submission of all blocks by manually setting last submitted height",
-			init: fillWithBlocks,
+			init: fillWithBlockData,
 			exec: func(ctx context.Context, t *testing.T, pb *PendingHeaders) {
 				pb.lastSubmittedHeight.Store(numBlocks)
 			},
@@ -52,7 +52,7 @@ func TestPendingBlocks(t *testing.T) {
 		},
 		{
 			name: "mock successful DA submission of all blocks by setting last submitted height using store",
-			init: fillWithBlocks,
+			init: fillWithBlockData,
 			exec: func(ctx context.Context, t *testing.T, pb *PendingHeaders) {
 				pb.lastSubmittedHeight.Store(pb.store.Height())
 			},
@@ -85,7 +85,7 @@ func newPendingBlocks(t *testing.T) *PendingHeaders {
 	return pendingBlocks
 }
 
-func fillWithBlocks(ctx context.Context, t *testing.T, pb *PendingHeaders) {
+func fillWithBlockData(ctx context.Context, t *testing.T, pb *PendingHeaders) {
 	for i := uint64(1); i <= numBlocks; i++ {
 		h, d := types.GetRandomBlock(i, 0)
 		require.NoError(t, pb.store.SaveBlockData(ctx, h, d, &types.Signature{}))

mempool/reaper.go

@@ -8,7 +8,7 @@ import (
 	cmtypes "github.com/cometbft/cometbft/types"
 
 	"github.com/cometbft/cometbft/libs/log"
-	seqGRPC "github.com/rollkit/go-sequencing/proxy/grpc"
+	"github.com/rollkit/go-sequencing/proxy/grpc"
 )
 
 // ReapInterval is the interval at which the reaper checks the mempool for transactions to reap.
@@ -22,16 +22,16 @@ const (
 type CListMempoolReaper struct {
 	mempool    Mempool
 	stopCh     chan struct{}
-	grpcClient *seqGRPC.Client
+	grpcClient *grpc.Client
 	ctx        context.Context
 	rollupId   []byte
 	submitted  map[cmtypes.TxKey]struct{}
 	mu         sync.RWMutex // Add a mutex to protect the submitted map
 	logger     log.Logger
 }
 
-// NewCListMempool initializes the mempool and sets up the gRPC client.
-func NewCListMempoolReaper(mempool Mempool, rollupId []byte, seqClient *seqGRPC.Client, logger log.Logger) *CListMempoolReaper {
+// NewCListMempoolReaper initializes the mempool and sets up the gRPC client.
+func NewCListMempoolReaper(mempool Mempool, rollupId []byte, seqClient *grpc.Client, logger log.Logger) *CListMempoolReaper {
 	return &CListMempoolReaper{
 		mempool:    mempool,
 		stopCh:     make(chan struct{}),
@@ -61,6 +61,7 @@ func (r *CListMempoolReaper) StartReaper() error {
 	return nil
 }
 
+// UpdateCommitedTxs removes the committed transactions from the submitted map.
 func (r *CListMempoolReaper) UpdateCommitedTxs(txs []cmtypes.Tx) {
 	r.mu.Lock() // Lock the mutex before modifying the map
 	defer r.mu.Unlock()

sequencing/mock/cmd/main.go

@@ -8,7 +8,7 @@ import (
 	"os/signal"
 	"syscall"
 
-	seqGRPC "github.com/rollkit/go-sequencing/proxy/grpc"
+	"github.com/rollkit/go-sequencing/proxy/grpc"
 	"github.com/rollkit/go-sequencing/test"
 )
 
@@ -25,7 +25,7 @@ func main() {
 
 	dummySequencer := test.NewDummySequencer()
 
-	srv := seqGRPC.NewServer(dummySequencer, dummySequencer, dummySequencer)
+	srv := grpc.NewServer(dummySequencer, dummySequencer, dummySequencer)
 	log.Printf("Listening on: %s:%s", "localhost", "50051")
 	if err := srv.Serve(lis); err != nil {
 		log.Fatal("error while serving:", err)