julienrbrt · May 6, 2026 · May 6, 2026 · May 6, 2026 · May 6, 2026 · May 6, 2026
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changes
 
+- Add max bytes contraints in simple solo sequnecer [#3312](https://github.com/evstack/ev-node/pull/3312)
 - Add support for otlp in execution/grpc. [#3300](https://github.com/evstack/ev-node/pull/3300)
 - Optimization of mutex usage in cache for reaper [#3286](https://github.com/evstack/ev-node/pull/3286)
 - Add Unix domain socket support for gRPC execution endpoints via `unix:///path/to/socket` [#3297](https://github.com/evstack/ev-node/pull/3297)

diff --git a/pkg/sequencers/common/errors.go b/pkg/sequencers/common/errors.go
@@ -0,0 +1,10 @@
+package common
+
+import "errors"
+
+var (
+	// ErrInvalidId is returned when the chain id is invalid
+	ErrInvalidID = errors.New("invalid chain id")
+	// ErrQueueFull is returned when the batch queue has reached its maximum size
+	ErrQueueFull = errors.New("sequencer queue full")
+)
diff --git a/pkg/sequencers/single/queue.go b/pkg/sequencers/single/queue.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/binary"
 	"encoding/hex"
-	"errors"
 	"fmt"
 	"strconv"
 	"sync"
@@ -14,12 +13,12 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	coresequencer "github.com/evstack/ev-node/core/sequencer"
+	"github.com/evstack/ev-node/pkg/sequencers/common"
 	"github.com/evstack/ev-node/pkg/store"
 	pb "github.com/evstack/ev-node/types/pb/evnode/v1"
 )
 
-// ErrQueueFull is returned when the batch queue has reached its maximum size
-var ErrQueueFull = errors.New("batch queue is full")
+var ErrQueueFull = common.ErrQueueFull
 
 // initialSeqNum is the starting sequence number for new queues.
 // It is set to the middle of the uint64 range to allow for both

diff --git a/pkg/sequencers/single/sequencer.go b/pkg/sequencers/single/sequencer.go
@@ -24,8 +24,7 @@ import (
 	"github.com/evstack/ev-node/types"
 )
 
-// ErrInvalidId is returned when the chain id is invalid
-var ErrInvalidId = errors.New("invalid chain id")
+var ErrInvalidId = seqcommon.ErrInvalidID
 
 // Catch-up state machine states
 const (

diff --git a/pkg/sequencers/solo/sequencer.go b/pkg/sequencers/solo/sequencer.go
@@ -3,7 +3,6 @@ package solo
 import (
 	"bytes"
 	"context"
-	"errors"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -12,12 +11,35 @@ import (
 
 	"github.com/evstack/ev-node/core/execution"
 	coresequencer "github.com/evstack/ev-node/core/sequencer"
+	"github.com/evstack/ev-node/pkg/sequencers/common"
 )
 
-var ErrInvalidID = errors.New("invalid chain id")
+var (
+	ErrInvalidID = common.ErrInvalidID
+	ErrQueueFull = common.ErrQueueFull
+)
+
+var (
+	emptyBatch        = &coresequencer.Batch{}
+	submitBatchResp   = &coresequencer.SubmitBatchTxsResponse{}
+	verifyBatchOKResp = &coresequencer.VerifyBatchResponse{Status: true}
+)
 
 var _ coresequencer.Sequencer = (*SoloSequencer)(nil)
 
+// Option configures a SoloSequencer.
+type Option func(*SoloSequencer)
+
+// WithMaxQueueBytes sets a soft cap on the sequencer's in-memory tx queue.
+// SubmitBatchTxs admits txs while the cap has room and returns ErrQueueFull
+// when the incoming batch would exceed it. A zero value (default) disables
+// the cap.
+func WithMaxQueueBytes(n uint64) Option {
+	return func(s *SoloSequencer) {
+		s.maxQueueBytes = n
+	}
+}
+
 // SoloSequencer is a single-leader sequencer without forced inclusion
 // support. It maintains a simple in-memory queue of mempool transactions and
 // produces batches on demand.
@@ -28,21 +50,38 @@ type SoloSequencer struct {
 
 	daHeight atomic.Uint64
 
-	mu    sync.Mutex
-	queue [][]byte
+	mu            sync.Mutex
+	queue         [][]byte
+	queueBytes    uint64
+	maxQueueBytes uint64
 }
 
 func NewSoloSequencer(
 	logger zerolog.Logger,
 	id []byte,
 	executor execution.Executor,
+	opts ...Option,
 ) *SoloSequencer {
-	return &SoloSequencer{
+	if executor == nil {
+		panic("solo: executor must not be nil")
+	}
+
+	s := &SoloSequencer{
 		logger:   logger,
 		id:       id,
 		executor: executor,
 		queue:    make([][]byte, 0),
 	}
+
+	for _, opt := range opts {
+		opt(s)
+	}
+
+	logger.Debug().
+		Uint64("max_queue_bytes", s.maxQueueBytes).
+		Msg("solo sequencer initialized")
+
+	return s
 }
 
 func (s *SoloSequencer) isValid(id []byte) bool {
@@ -55,14 +94,38 @@ func (s *SoloSequencer) SubmitBatchTxs(ctx context.Context, req coresequencer.Su
 	}
 
 	if req.Batch == nil || len(req.Batch.Transactions) == 0 {
-		return &coresequencer.SubmitBatchTxsResponse{}, nil
+		return submitBatchResp, nil
 	}
 
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	if s.maxQueueBytes == 0 {
+		s.queue = append(s.queue, req.Batch.Transactions...)
+		return submitBatchResp, nil
+	}
+
+	// All-or-nothing: if the whole incoming batch doesn't fit, reject
+	// it untouched. Partial admission would force the caller (e.g.
+	// the reaper bridging executor mempool → sequencer) to reason
+	// about which prefix was admitted and re-feed only the suffix on
+	// retry, which it doesn't currently do — leading to duplicate-tx
+	// resubmission on each retry. Rejecting the whole batch lets the
+	// reaper just retry with the same batch later when the queue has
+	// drained.
+	var batchBytes uint64
+	for _, tx := range req.Batch.Transactions {
+		batchBytes += uint64(len(tx))
+	}
+
+	if s.queueBytes+batchBytes > s.maxQueueBytes {
+		return submitBatchResp, ErrQueueFull
+	}
+
 	s.queue = append(s.queue, req.Batch.Transactions...)
-	return &coresequencer.SubmitBatchTxsResponse{}, nil
+	s.queueBytes += batchBytes
+
+	return submitBatchResp, nil
 }
 
 func (s *SoloSequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextBatchRequest) (*coresequencer.GetNextBatchResponse, error) {
@@ -73,11 +136,12 @@ func (s *SoloSequencer) GetNextBatch(ctx context.Context, req coresequencer.GetN
 	s.mu.Lock()
 	txs := s.queue
 	s.queue = nil
+	s.queueBytes = 0
 	s.mu.Unlock()
 
 	if len(txs) == 0 {
 		return &coresequencer.GetNextBatchResponse{
-			Batch:     &coresequencer.Batch{},
+			Batch:     emptyBatch,
 			Timestamp: time.Now().UTC(),
 			BatchData: req.LastBatchData,
 		}, nil
@@ -94,32 +158,41 @@ func (s *SoloSequencer) GetNextBatch(ctx context.Context, req coresequencer.GetN
 	filterStatuses, err := s.executor.FilterTxs(ctx, txs, req.MaxBytes, maxGas, false)
 	if err != nil {
 		s.logger.Warn().Err(err).Msg("failed to filter transactions, proceeding with unfiltered")
-		filterStatuses = make([]execution.FilterStatus, len(txs))
-		for i := range filterStatuses {
-			filterStatuses[i] = execution.FilterOK
-		}
+		return &coresequencer.GetNextBatchResponse{
+			Batch:     &coresequencer.Batch{Transactions: txs},
+			Timestamp: time.Now().UTC(),
+			BatchData: req.LastBatchData,
+		}, nil
 	}
 
-	var validTxs [][]byte
+	write := 0
 	var postponedTxs [][]byte
 	for i, status := range filterStatuses {
 		switch status {
 		case execution.FilterOK:
-			validTxs = append(validTxs, txs[i])
+			txs[write] = txs[i]
+			write++
 		case execution.FilterPostpone:
 			postponedTxs = append(postponedTxs, txs[i])
-		case execution.FilterRemove:
 		}
 	}
 
 	if len(postponedTxs) > 0 {
 		s.mu.Lock()
 		s.queue = append(postponedTxs, s.queue...)
+		// Postponed txs were already in the queue's byte count when
+		// SubmitBatchTxs admitted them. We zeroed queueBytes on drain
+		// above, so re-queuing requires re-counting whatever survived.
+		var bytes uint64
+		for _, tx := range postponedTxs {
+			bytes += uint64(len(tx))
+		}
+		s.queueBytes += bytes
 		s.mu.Unlock()
 	}
 
 	return &coresequencer.GetNextBatchResponse{
-		Batch:     &coresequencer.Batch{Transactions: validTxs},
+		Batch:     &coresequencer.Batch{Transactions: txs[:write]},
 		Timestamp: time.Now().UTC(),
 		BatchData: req.LastBatchData,
 	}, nil
@@ -130,7 +203,7 @@ func (s *SoloSequencer) VerifyBatch(ctx context.Context, req coresequencer.Verif
 		return nil, ErrInvalidID
 	}
 
-	return &coresequencer.VerifyBatchResponse{Status: true}, nil
+	return verifyBatchOKResp, nil
 }
 
 func (s *SoloSequencer) SetDAHeight(height uint64) {

diff --git a/pkg/sequencers/solo/sequencer_test.go b/pkg/sequencers/solo/sequencer_test.go
@@ -217,3 +217,91 @@ func TestSoloSequencer_DAHeight(t *testing.T) {
 	seq.SetDAHeight(42)
 	assert.Equal(t, uint64(42), seq.GetDAHeight())
 }
+
+func TestSoloSequencer_SubmitBatchTxs_QueueFull(t *testing.T) {
+	seq := NewSoloSequencer(
+		zerolog.Nop(),
+		[]byte("test"),
+		createDefaultMockExecutor(t),
+		WithMaxQueueBytes(10),
+	)
+
+	_, err := seq.SubmitBatchTxs(context.Background(), coresequencer.SubmitBatchTxsRequest{
+		Id:    []byte("test"),
+		Batch: &coresequencer.Batch{Transactions: [][]byte{make([]byte, 6)}},
+	})
+	require.NoError(t, err)
+
+	_, err = seq.SubmitBatchTxs(context.Background(), coresequencer.SubmitBatchTxsRequest{
+		Id:    []byte("test"),
+		Batch: &coresequencer.Batch{Transactions: [][]byte{make([]byte, 5)}},
+	})
+	assert.ErrorIs(t, err, ErrQueueFull)
+}
+
+func TestSoloSequencer_SubmitBatchTxs_QueueFull_AllOrNothing(t *testing.T) {
+	seq := NewSoloSequencer(
+		zerolog.Nop(),
+		[]byte("test"),
+		createDefaultMockExecutor(t),
+		WithMaxQueueBytes(5),
+	)
+
+	_, err := seq.SubmitBatchTxs(context.Background(), coresequencer.SubmitBatchTxsRequest{
+		Id:    []byte("test"),
+		Batch: &coresequencer.Batch{Transactions: [][]byte{{1}, {2}, {3}}},
+	})
+	require.NoError(t, err)
+
+	_, err = seq.SubmitBatchTxs(context.Background(), coresequencer.SubmitBatchTxsRequest{
+		Id:    []byte("test"),
+		Batch: &coresequencer.Batch{Transactions: [][]byte{{4}, {5}, {6}}},
+	})
+	assert.ErrorIs(t, err, ErrQueueFull)
+
+	assert.Len(t, seq.queue, 3, "queue should contain only the first batch")
+}
+
+func TestSoloSequencer_SubmitBatchTxs_DrainReleasesCapacity(t *testing.T) {
+	seq := NewSoloSequencer(
+		zerolog.Nop(),
+		[]byte("test"),
+		createDefaultMockExecutor(t),
+		WithMaxQueueBytes(10),
+	)
+
+	_, err := seq.SubmitBatchTxs(context.Background(), coresequencer.SubmitBatchTxsRequest{
+		Id:    []byte("test"),
+		Batch: &coresequencer.Batch{Transactions: [][]byte{make([]byte, 10)}},
+	})
+	require.NoError(t, err)
+
+	_, err = seq.SubmitBatchTxs(context.Background(), coresequencer.SubmitBatchTxsRequest{
+		Id:    []byte("test"),
+		Batch: &coresequencer.Batch{Transactions: [][]byte{{1}}},
+	})
+	assert.ErrorIs(t, err, ErrQueueFull)
+
+	_, err = seq.GetNextBatch(context.Background(), coresequencer.GetNextBatchRequest{Id: []byte("test")})
+	require.NoError(t, err)
+
+	_, err = seq.SubmitBatchTxs(context.Background(), coresequencer.SubmitBatchTxsRequest{
+		Id:    []byte("test"),
+		Batch: &coresequencer.Batch{Transactions: [][]byte{{1}}},
+	})
+	assert.NoError(t, err, "submit should succeed after queue is drained")
+}
+
+func TestSoloSequencer_SubmitBatchTxs_UnboundedByDefault(t *testing.T) {
+	seq := newTestSequencer(t)
+
+	bigTx := make([]byte, 1024*1024)
+	for i := 0; i < 10; i++ {
+		_, err := seq.SubmitBatchTxs(context.Background(), coresequencer.SubmitBatchTxsRequest{
+			Id:    []byte("test"),
+			Batch: &coresequencer.Batch{Transactions: [][]byte{bigTx}},
+		})
+		require.NoError(t, err)
+	}
+	assert.Len(t, seq.queue, 10)
+}