add persistence to pendingTxs in the sequencer

tac0turtle · Apr 18, 2025 · Apr 9, 2025 · Apr 9, 2025 · Apr 9, 2025 · Apr 9, 2025
commit fa00b6c0af0062ec4990ebbc3092bedf61b0b6ce
diff --git a/rollups/evm/based/main.go b/rollups/evm/based/main.go
@@ -144,13 +144,19 @@ func NewExtendedRunNodeCmd(ctx context.Context) *cobra.Command {
 			nsBytes, err := hex.DecodeString(basedNamespace)
 			basedDALC := da.NewDAClient(basedDA, basedGasPrice, basedGasMultiplier, nsBytes, nil, logger)
 
+			datastore, err := store.NewDefaultKVStore(nodeConfig.RootDir, nodeConfig.DBPath, "based")
+			if err != nil {
+				panic(err)
+			}
+
 			sequencer, err := based.NewSequencer(
 				logger,
 				basedDA,
 				basedDALC,
 				[]byte("rollkit-test"),
 				basedStartHeight,
 				basedMaxHeightDrift,
+				datastore,
 			)
 			if err != nil {
 				return fmt.Errorf("failed to create based sequencer: %w", err)
@@ -170,11 +176,6 @@ func NewExtendedRunNodeCmd(ctx context.Context) *cobra.Command {
 				panic(err)
 			}
 
-			datastore, err := store.NewDefaultKVStore(nodeConfig.RootDir, nodeConfig.DBPath, "based")
-			if err != nil {
-				panic(err)
-			}
-
 			p2pClient, err := p2p.NewClient(nodeConfig, "based", nodeKey, datastore, logger, p2p.NopMetrics())
 			if err != nil {
 				panic(err)

diff --git a/sequencers/based/go.mod b/sequencers/based/go.mod
@@ -10,10 +10,10 @@ replace (
 
 require (
 	cosmossdk.io/log v1.5.0
+	github.com/ipfs/go-datastore v0.7.0
 	github.com/rollkit/rollkit/core v0.0.0-20250312114929-104787ba1a4c
 	github.com/rollkit/rollkit/da v0.0.0-00010101000000-000000000000
 	github.com/stretchr/testify v1.10.0
-	github.com/test-go/testify v1.1.4
 )
 
 require (
@@ -22,14 +22,13 @@ require (
 	github.com/cloudwego/base64x v0.1.4 // indirect
 	github.com/cloudwego/iasm v0.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/google/uuid v1.6.0 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
-	github.com/kr/text v0.2.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/rs/zerolog v1.33.0 // indirect
-	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	golang.org/x/arch v0.0.0-20210923205945-b76863e36670 // indirect
 	golang.org/x/sys v0.31.0 // indirect

diff --git a/sequencers/based/go.sum b/sequencers/based/go.sum
@@ -10,12 +10,17 @@ github.com/cloudwego/base64x v0.1.4/go.mod h1:0zlkT4Wn5C6NdauXdJRhSKRlJvmclQ1hhJ
 github.com/cloudwego/iasm v0.2.0 h1:1KNIy1I1H9hNNFEEH3DVnI4UujN+1zjpuk6gwHLTssg=
 github.com/cloudwego/iasm v0.2.0/go.mod h1:8rXZaNYT2n95jn+zTI1sDr+IgcD2GVs0nlbbQPiEFhY=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
-github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/ipfs/go-datastore v0.8.2 h1:Jy3wjqQR6sg/LhyY0NIePZC3Vux19nLtg7dx0TVqr6U=
+github.com/ipfs/go-datastore v0.8.2/go.mod h1:W+pI1NsUsz3tcsAACMtfC+IZdnQTnC/7VfPoJBQuts0=
+github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ6cOk=
+github.com/ipfs/go-detect-race v0.0.1/go.mod h1:8BNT7shDZPo99Q74BpGMK+4D8Mn4j46UU0LZ723meps=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.9 h1:66ze0taIn2H33fBvCkXuv9BmCwDfafmiIVpKV9kKGuY=
 github.com/klauspost/cpuid/v2 v2.2.9/go.mod h1:rqkxqrZ1EhYM9G+hXH7YdowN5R5RGN6NK4QwQ3WMXF8=
@@ -51,8 +56,6 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-github.com/test-go/testify v1.1.4 h1:Tf9lntrKUMHiXQ07qBScBTSA0dhYQlu83hswqelv1iE=
-github.com/test-go/testify v1.1.4/go.mod h1:rH7cfJo/47vWGdi4GPj16x3/t1xGOj2YxzmNQzk2ghU=
 github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 golang.org/x/arch v0.0.0-20210923205945-b76863e36670 h1:18EFjUmQOcUvxNYSkA6jO9VAiXCnxFY6NyDX0bHDmkU=

diff --git a/sequencers/based/sequencer.go b/sequencers/based/sequencer.go
@@ -3,12 +3,14 @@ package based
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"time"
 
 	"cosmossdk.io/log"
 
+	datastore "github.com/ipfs/go-datastore"
 	coreda "github.com/rollkit/rollkit/core/da"
 	coresequencer "github.com/rollkit/rollkit/core/sequencer"
 )
@@ -20,76 +22,97 @@ var (
 	batchTime         = 1 * time.Second
 )
 
-const DefaultMaxBlobSize uint64 = 1_500_000
+const (
+	DefaultMaxBlobSize uint64 = 1_500_000
+	dsPendingTxsKey           = "/sequencer/pendingTxs"
+)
 
-var ErrInvalidRollupId = errors.New("invalid rollup id")
-var ErrInvalidMaxBytes = errors.New("invalid max bytes")
+var (
+	ErrInvalidRollupId = errors.New("invalid rollup id")
+	ErrInvalidMaxBytes = errors.New("invalid max bytes")
+)
 
 type TxsWithTimestamp struct {
 	Txs       [][]byte
 	IDs       [][]byte
 	Timestamp time.Time
 }
 
-type TxsWithTimestamps []TxsWithTimestamp
-
-// Push adds a new TxsWithTimestamp to the TxsWithTimestamps slice.
-func (t *TxsWithTimestamps) Push(txs [][]byte, ids [][]byte, timestamp time.Time) {
-	*t = append(*t, TxsWithTimestamp{Txs: txs, IDs: ids, Timestamp: timestamp})
+type PersistentPendingTxs struct {
+	store datastore.Batching
+	list  []TxsWithTimestamp
 }
 
-// Pull removes and returns the first TxsWithTimestamp from the TxsWithTimestamps slice.
-// If the slice is empty, it returns nil and a zero-value TxsWithTimestamp.
-func (t *TxsWithTimestamps) Pull() (TxsWithTimestamp, bool) {
-	if len(*t) == 0 {
-		return TxsWithTimestamp{}, false
+func NewPersistentPendingTxs(store datastore.Batching) (*PersistentPendingTxs, error) {
+	pt := &PersistentPendingTxs{store: store, list: []TxsWithTimestamp{}}
+	if err := pt.Load(); err != nil && err != datastore.ErrNotFound {
+		return nil, err
 	}
-	first := (*t)[0]
-	*t = (*t)[1:]
-	return first, true
+	return pt, nil
 }
 
-// PopUpToMaxBytes pops transactions from TxsWithTimestamps up to the specified maxBytes.
-// It returns the popped transactions and the total size in bytes.
-func (t *TxsWithTimestamps) PopUpToMaxBytes(maxBytes uint64) ([][]byte, [][]byte, uint64, time.Time) {
+func (pt *PersistentPendingTxs) Push(txs [][]byte, ids [][]byte, timestamp time.Time) error {
+	pt.list = append(pt.list, TxsWithTimestamp{Txs: txs, IDs: ids, Timestamp: timestamp})
+	return pt.Save()
+}
+
+func (pt *PersistentPendingTxs) PopUpToMaxBytes(maxBytes uint64) ([][]byte, [][]byte, uint64, time.Time) {
 	var (
 		poppedTxs [][]byte
 		ids       [][]byte
 		totalSize uint64
-		timestamp time.Time
+		timestamp time.Time = time.Now()
 	)
-	// set the timestamp to sequencer time, until we find a timestamp from the base layer
-	timestamp = time.Now()
 
-	for len(*t) > 0 {
-		first := (*t)[0]
+	for len(pt.list) > 0 {
+		first := pt.list[0]
 		timestamp = first.Timestamp
 		for i, tx := range first.Txs {
 			txSize := uint64(len(tx))
 			if totalSize+txSize > maxBytes {
-				// Push remaining transactions to the front of the queue
-				*t = append([]TxsWithTimestamp{{Txs: first.Txs[i:], IDs: first.IDs[i:], Timestamp: first.Timestamp}}, (*t)...)
+				pt.list[0] = TxsWithTimestamp{
+					Txs:       first.Txs[i:],
+					IDs:       first.IDs[i:],
+					Timestamp: first.Timestamp,
+				}
+				pt.Save()
 				return poppedTxs, ids, totalSize, timestamp
 			}
 			poppedTxs = append(poppedTxs, tx)
 			ids = append(ids, first.IDs[i])
 			totalSize += txSize
 		}
-		*t = (*t)[1:]
+		pt.list = pt.list[1:]
 	}
-
+	pt.Save()
 	return poppedTxs, ids, totalSize, timestamp
 }
 
+func (pt *PersistentPendingTxs) Save() error {
+	data, err := json.Marshal(pt.list)
+	if err != nil {
+		return err
+	}
+	return pt.store.Put(context.Background(), datastore.NewKey(dsPendingTxsKey), data)
+}
+
+func (pt *PersistentPendingTxs) Load() error {
+	data, err := pt.store.Get(context.Background(), datastore.NewKey(dsPendingTxsKey))
+	if err != nil {
+		return err
+	}
+	return json.Unmarshal(data, &pt.list)
+}
+
 var _ coresequencer.Sequencer = &Sequencer{}
 
 type Sequencer struct {
 	logger         log.Logger
 	maxHeightDrift uint64
 	rollupId       []byte
-	da             coreda.DA
+	DA             coreda.DA
 	dalc           coreda.Client
-	pendingTxs     TxsWithTimestamps
+	pendingTxs     *PersistentPendingTxs
 	daStartHeight  uint64
 }
 
@@ -100,17 +123,21 @@ func NewSequencer(
 	rollupId []byte,
 	daStartHeight uint64,
 	maxHeightDrift uint64,
+	ds datastore.Batching,
 ) (*Sequencer, error) {
-	s := &Sequencer{
+	pending, err := NewPersistentPendingTxs(ds)
+	if err != nil {
+		return nil, err
+	}
+	return &Sequencer{
 		logger:         logger,
 		maxHeightDrift: maxHeightDrift,
 		rollupId:       rollupId,
-		da:             da,
+		DA:             da,
 		dalc:           dalc,
 		daStartHeight:  daStartHeight,
-		pendingTxs:     TxsWithTimestamps{},
-	}
-	return s, nil
+		pendingTxs:     pending,
+	}, nil
 }
 
 // AddToPendingTxs adds transactions to the pending queue.
@@ -219,13 +246,13 @@ func (s *Sequencer) VerifyBatch(ctx context.Context, req coresequencer.VerifyBat
 		return nil, fmt.Errorf("failed to get namespace: %w", err)
 	}
 	// get the proofs
-	proofs, err := s.da.GetProofs(ctx, req.BatchData, namespace)
+	proofs, err := s.DA.GetProofs(ctx, req.BatchData, namespace)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get proofs: %w", err)
 	}
 
 	// verify the proof
-	valid, err := s.da.Validate(ctx, req.BatchData, proofs, namespace)
+	valid, err := s.DA.Validate(ctx, req.BatchData, proofs, namespace)
 	if err != nil {
 		return nil, fmt.Errorf("failed to validate proof: %w", err)
 	}