add persistence to reaper and avoid submitting duplicate transactions

tac0turtle · Apr 18, 2025 · Apr 9, 2025 · Apr 9, 2025 · Apr 9, 2025 · Apr 9, 2025
commit e6baffb36527b2f14e3b5133f315fc6eb0e0a617
diff --git a/block/reaper.go b/block/reaper.go
@@ -2,38 +2,40 @@ package block
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"time"
 
 	"cosmossdk.io/log"
+	ds "github.com/ipfs/go-datastore"
 	"github.com/rollkit/go-sequencing"
 	coreexecutor "github.com/rollkit/rollkit/core/execution"
 	coresequencer "github.com/rollkit/rollkit/core/sequencer"
 )
 
-// Reaper is responsible for periodically pulling transactions from the executor
-// and submitting them to the sequencer.
 type Reaper struct {
 	exec      coreexecutor.Executor
 	sequencer coresequencer.Sequencer
 	chainID   string
 	interval  time.Duration
 	logger    log.Logger
 	ctx       context.Context
+	seenStore ds.Batching
 }
 
-// NewReaper creates a new Reaper instance.
-func NewReaper(ctx context.Context, exec coreexecutor.Executor, sequencer coresequencer.Sequencer, chainID string, interval time.Duration, logger log.Logger) *Reaper {
+// NewReaper creates a new Reaper instance with persistent seenTx storage.
+func NewReaper(ctx context.Context, exec coreexecutor.Executor, sequencer coresequencer.Sequencer, chainID string, interval time.Duration, logger log.Logger, store ds.Batching) *Reaper {
 	return &Reaper{
 		exec:      exec,
 		sequencer: sequencer,
 		chainID:   chainID,
 		interval:  interval,
 		logger:    logger,
 		ctx:       ctx,
+		seenStore: store,
 	}
 }
 
-// Start runs the reaper loop which pulls and submits transactions at a fixed interval.
 func (r *Reaper) Start() {
 	ticker := time.NewTicker(r.interval)
 	defer ticker.Stop()
@@ -51,29 +53,55 @@ func (r *Reaper) Start() {
 	}
 }
 
-// SubmitTxs pulls transactions from the executor and submits them to the sequencer.
 func (r *Reaper) SubmitTxs() {
 	txs, err := r.exec.GetTxs(r.ctx)
 	if err != nil {
 		r.logger.Error("Reaper failed to get txs from executor", "error", err)
 		return
 	}
 
-	if len(txs) == 0 {
-		r.logger.Debug("Reaper found no txs to submit")
+	var newTxs [][]byte
+	for _, tx := range txs {
+		txHash := hashTx(tx)
+		key := ds.NewKey(txHash)
+		has, err := r.seenStore.Has(r.ctx, key)
+		if err != nil {
+			r.logger.Error("Failed to check seenStore", "error", err)
+			continue
+		}
+		if !has {
+			newTxs = append(newTxs, tx)
+		}
+	}
+
+	if len(newTxs) == 0 {
+		r.logger.Debug("Reaper found no new txs to submit")
 		return
 	}
 
-	r.logger.Debug("Reaper submitting txs to sequencer", "txCount", len(txs))
+	r.logger.Debug("Reaper submitting txs to sequencer", "txCount", len(newTxs))
 
 	_, err = r.sequencer.SubmitRollupBatchTxs(r.ctx, coresequencer.SubmitRollupBatchTxsRequest{
 		RollupId: sequencing.RollupId(r.chainID),
-		Batch:    &coresequencer.Batch{Transactions: txs},
+		Batch:    &coresequencer.Batch{Transactions: newTxs},
 	})
 	if err != nil {
 		r.logger.Error("Reaper failed to submit txs to sequencer", "error", err)
 		return
 	}
 
+	for _, tx := range newTxs {
+		txHash := hashTx(tx)
+		key := ds.NewKey(txHash)
+		if err := r.seenStore.Put(r.ctx, key, []byte{1}); err != nil {
+			r.logger.Error("Failed to persist seen tx", "txHash", txHash, "error", err)
+		}
+	}
+
 	r.logger.Debug("Reaper successfully submitted txs")
 }
+
+func hashTx(tx []byte) string {
+	hash := sha256.Sum256(tx)
+	return hex.EncodeToString(hash[:])
+}
diff --git a/block/reaper_test.go b/block/reaper_test.go
@@ -2,10 +2,14 @@ package block
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"testing"
 	"time"
 
 	"cosmossdk.io/log"
+	ds "github.com/ipfs/go-datastore"
+	dsync "github.com/ipfs/go-datastore/sync"
 	execution "github.com/rollkit/rollkit/core/execution"
 	sequencer "github.com/rollkit/rollkit/core/sequencer"
 )
@@ -16,14 +20,16 @@ func TestReaper_SubmitTxs_Success(t *testing.T) {
 
 	exec := execution.NewDummyExecutor()
 	seq := sequencer.NewDummySequencer()
+	store := dsync.MutexWrap(ds.NewMapDatastore())
 	logger := log.NewNopLogger()
 	chainID := "test-chain"
 	interval := 100 * time.Millisecond
 
-	reaper := NewReaper(ctx, exec, seq, chainID, interval, logger)
+	reaper := NewReaper(ctx, exec, seq, chainID, interval, logger, store)
 
 	// Inject a transaction
-	exec.InjectTx([]byte("tx1"))
+	tx := []byte("tx1")
+	exec.InjectTx(tx)
 
 	// Run once and ensure transaction is submitted
 	reaper.SubmitTxs()
@@ -35,6 +41,17 @@ func TestReaper_SubmitTxs_Success(t *testing.T) {
 	if len(res.Batch.Transactions) != 1 {
 		t.Fatalf("expected 1 transaction submitted, got %d", len(res.Batch.Transactions))
 	}
+
+	// Run again, should not resubmit
+	reaper.SubmitTxs()
+
+	res, err = seq.GetNextBatch(ctx, sequencer.GetNextBatchRequest{RollupId: []byte(chainID)})
+	if err != nil {
+		t.Fatalf("expected success, got error: %v", err)
+	}
+	if len(res.Batch.Transactions) != 1 {
+		t.Fatalf("expected still 1 transaction, got %d", len(res.Batch.Transactions))
+	}
 }
 
 func TestReaper_SubmitTxs_NoTxs(t *testing.T) {
@@ -43,18 +60,19 @@ func TestReaper_SubmitTxs_NoTxs(t *testing.T) {
 
 	exec := execution.NewDummyExecutor()
 	seq := sequencer.NewDummySequencer()
+	store := dsync.MutexWrap(ds.NewMapDatastore())
 	logger := log.NewNopLogger()
 	chainID := "test-chain"
 	interval := 100 * time.Millisecond
 
-	reaper := NewReaper(ctx, exec, seq, chainID, interval, logger)
+	reaper := NewReaper(ctx, exec, seq, chainID, interval, logger, store)
 
 	// Run once and ensure nothing is submitted
 	reaper.SubmitTxs()
 
-	res, err := seq.GetNextBatch(ctx, sequencer.GetNextBatchRequest{RollupId: []byte(chainID)})
+	_, err := seq.GetNextBatch(ctx, sequencer.GetNextBatchRequest{RollupId: []byte(chainID)})
 	if err == nil {
-		t.Fatalf("expected error due to no batch submitted, got batch: %+v", res)
+		t.Fatalf("expected error due to no batch submitted")
 	}
 }
 
@@ -64,20 +82,53 @@ func TestReaper_SubmitTxs_InvalidChainID(t *testing.T) {
 
 	exec := execution.NewDummyExecutor()
 	seq := sequencer.NewDummySequencer()
+	store := dsync.MutexWrap(ds.NewMapDatastore())
 	logger := log.NewNopLogger()
 	chainID := "test-chain"
 	interval := 100 * time.Millisecond
 
-	reaper := NewReaper(ctx, exec, seq, chainID, interval, logger)
+	reaper := NewReaper(ctx, exec, seq, chainID, interval, logger, store)
 
 	// Inject a transaction
 	exec.InjectTx([]byte("tx1"))
 
-	// Override chain ID to invalid one (different in GetNextBatch call)
+	// Run with correct ID to store the tx
 	reaper.SubmitTxs()
 
-	_, err := seq.GetNextBatch(ctx, sequencer.GetNextBatchRequest{RollupId: []byte("different-chain")})
+	_, err := seq.GetNextBatch(ctx, sequencer.GetNextBatchRequest{RollupId: []byte("wrong-id")})
 	if err == nil {
 		t.Fatal("expected error due to invalid rollup ID")
 	}
 }
+
+func TestReaper_TxPersistence_AcrossRestarts(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	exec := execution.NewDummyExecutor()
+	seq := sequencer.NewDummySequencer()
+	store := dsync.MutexWrap(ds.NewMapDatastore())
+	logger := log.NewNopLogger()
+	chainID := "test-chain"
+	interval := 100 * time.Millisecond
+
+	// First reaper instance
+	reaper := NewReaper(ctx, exec, seq, chainID, interval, logger, store)
+	tx := []byte("tx-persist")
+	exec.InjectTx(tx)
+	reaper.SubmitTxs()
+
+	// Create a new reaper instance simulating a restart
+	reaper2 := NewReaper(ctx, exec, seq, chainID, interval, logger, store)
+	exec.InjectTx(tx) // re-inject the same tx
+
+	// Should not submit it again
+	reaper2.SubmitTxs()
+
+	txHash := sha256.Sum256(tx)
+	key := ds.NewKey(hex.EncodeToString(txHash[:]))
+	has, err := store.Has(ctx, key)
+	if err != nil || !has {
+		t.Fatalf("expected tx to be marked as seen, err=%v", err)
+	}
+}
diff --git a/node/full.go b/node/full.go
@@ -103,6 +103,7 @@ func newFullNode(
 		genesis.ChainID,
 		nodeConfig.Node.BlockTime.Duration,
 		logger.With("module", "Reaper"),
+		mainKV,
 	)
 
 	blockManager, err := initBlockManager(