save txs to file

Author: udpatil

config/settings.go

@@ -24,6 +24,7 @@ type Settings struct {
 	Prewarm          bool     `json:"prewarm,omitempty"`
 	RampUp           bool     `json:"rampUp,omitempty"`
 	ReportPath       string   `json:"reportPath,omitempty"`
+	TxsDir           string   `json:"txsDir,omitempty"`
 }
 
 // DefaultSettings returns the default configuration values
@@ -41,6 +42,7 @@ func DefaultSettings() Settings {
 		Prewarm:          false,
 		RampUp:           false,
 		ReportPath:       "",
+		TxsDir:           "",
 	}
 }
 
@@ -60,6 +62,7 @@ func InitializeViper(cmd *cobra.Command) error {
 		"workers":          "workers",
 		"rampUp":           "ramp-up",
 		"reportPath":       "report-path",
+		"txsDir":           "txs-dir",
 	}
 
 	for viperKey, flagName := range flagBindings {
@@ -82,6 +85,7 @@ func InitializeViper(cmd *cobra.Command) error {
 	viper.SetDefault("workers", defaults.Workers)
 	viper.SetDefault("rampUp", defaults.RampUp)
 	viper.SetDefault("reportPath", defaults.ReportPath)
+	viper.SetDefault("txsDir", defaults.TxsDir)
 	return nil
 }
 
@@ -120,5 +124,6 @@ func ResolveSettings() Settings {
 		Prewarm:          viper.GetBool("prewarm"),
 		RampUp:           viper.GetBool("rampUp"),
 		ReportPath:       viper.GetString("reportPath"),
+		TxsDir:           viper.GetString("txsDir"),
 	}
 }

main.go

@@ -65,6 +65,7 @@ func init() {
 	rootCmd.Flags().String("metricsListenAddr", "0.0.0.0:9090", "The ip:port on which to export prometheus metrics.")
 	rootCmd.Flags().Bool("ramp-up", false, "Ramp up loadtest")
 	rootCmd.Flags().String("report-path", "", "Path to save the report")
+	rootCmd.Flags().String("txs-dir", "", "Path to save the transactions")
 
 	// Initialize Viper with proper error handling
 	if err := config.InitializeViper(rootCmd); err != nil {
@@ -169,12 +170,6 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command, args []string) error {
 			sharedLimiter = rate.NewLimiter(rate.Inf, 1)
 		}
 
-		// Create the sender from the config struct
-		snd, err := sender.NewShardedSender(cfg, settings.BufferSize, settings.Workers, sharedLimiter)
-		if err != nil {
-			return fmt.Errorf("failed to create sender: %w", err)
-		}
-
 		// Create and start block collector if endpoints are available
 		var blockCollector *stats.BlockCollector
 		if len(cfg.Endpoints) > 0 && settings.TrackBlocks {
@@ -207,6 +202,12 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command, args []string) error {
 			})
 		}
 
+		// Create the sender from the config struct
+		snd, err := sender.NewShardedSender(cfg, settings.BufferSize, settings.Workers, sharedLimiter)
+		if err != nil {
+			return fmt.Errorf("failed to create sender: %w", err)
+		}
+
 		// Enable dry-run mode in sender if specified
 		if settings.DryRun {
 			snd.SetDryRun(true)
@@ -225,7 +226,13 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command, args []string) error {
 		snd.SetStatsCollector(collector, logger)
 
 		// Create dispatcher
-		dispatcher := sender.NewDispatcher(gen, snd)
+		var dispatcher *sender.Dispatcher
+		if settings.TxsDir != "" {
+			writer := sender.NewTxsWriter(10_000_000, settings.TxsDir)
+			dispatcher = sender.NewDispatcher(gen, writer)
+		} else {
+			dispatcher = sender.NewDispatcher(gen, snd)
+		}
 
 		// Set statistics collector for dispatcher
 		dispatcher.SetStatsCollector(collector)
@@ -239,10 +246,11 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command, args []string) error {
 			log.Printf("📝 Prewarm mode: Accounts will be prewarmed")
 		}
 
-		// Start the sender (starts all workers)
-		s.SpawnBgNamed("sender", func() error { return snd.Run(ctx) })
-		log.Printf("✅ Connected to %d endpoints", snd.GetNumShards())
-
+		if settings.TxsDir == "" {
+			// Start the sender (starts all workers)
+			s.SpawnBgNamed("sender", func() error { return snd.Run(ctx) })
+			log.Printf("✅ Connected to %d endpoints", snd.GetNumShards())
+		}
 		// Perform prewarming if enabled (before starting logger to avoid logging prewarm transactions)
 		if settings.Prewarm {
 			if err := dispatcher.Prewarm(ctx); err != nil {

profiles/conflict.json

@@ -0,0 +1,30 @@
+{
+    "chainId": 713714,
+    "seiChainId": "sei-chain",
+    "endpoints": [
+      "http://127.0.0.1:8545"
+    ],
+    "accounts": {
+      "count": 5000,
+      "newAccountRate": 0.0
+    },
+    "scenarios": [
+      {
+        "name": "ERC20Conflict",
+        "weight": 1
+      }
+    ],
+    "settings": {
+      "workers": 1,
+      "tps": 0,
+      "statsInterval": "5s",
+      "bufferSize": 1000,
+      "dryRun": false,
+      "debug": false,
+      "trackReceipts": false,
+      "trackBlocks": false,
+      "trackUserLatency": false,
+      "prewarm": false,
+      "rampUp": false
+    }
+  }

sender/writer.go

@@ -0,0 +1,86 @@
+package sender
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"os"
+	"path/filepath"
+
+	"github.com/sei-protocol/sei-load/types"
+)
+
+// implements `Send`
+
+type TxsWriter struct {
+	gasPerBlock uint64
+	nextHeight  uint64
+	txsDir      string
+
+	bufferGas uint64
+	txBuffer  []*types.LoadTx
+}
+
+func NewTxsWriter(gasPerBlock uint64, txsDir string) *TxsWriter {
+	// what height to start at?
+	return &TxsWriter{
+		gasPerBlock: gasPerBlock,
+		nextHeight:  1,
+		txsDir:      txsDir,
+
+		bufferGas: 0,
+		txBuffer:  make([]*types.LoadTx, 0),
+	}
+}
+
+// Send writes the transaction to the writer
+func (w *TxsWriter) Send(ctx context.Context, tx *types.LoadTx) error {
+	// if bwe would exceed gasPerBlock, flush
+	if w.bufferGas+tx.EthTx.Gas() > w.gasPerBlock {
+		if err := w.Flush(); err != nil {
+			return err
+		}
+	}
+
+	// add to buffer
+	w.txBuffer = append(w.txBuffer, tx)
+	w.bufferGas += tx.EthTx.Gas()
+	return nil
+}
+
+type TxWriteData struct {
+	TxPayloads [][]byte `json:"tx_payloads"`
+}
+
+func (w *TxsWriter) Flush() error {
+	defer func() {
+		// clear buffer and reset bufferGas and increment nextHeight
+		w.txBuffer = make([]*types.LoadTx, 0)
+		w.bufferGas = 0
+		w.nextHeight++
+	}()
+	// write to dir `~/load_txs`
+	// make dir if it doesn't exist
+	os.MkdirAll(w.txsDir, 0755)
+	txsFile := filepath.Join(w.txsDir, fmt.Sprintf("%d_txs.json", w.nextHeight))
+	txData := TxWriteData{
+		TxPayloads: make([][]byte, 0),
+	}
+	for _, tx := range w.txBuffer {
+		txData.TxPayloads = append(txData.TxPayloads, tx.Payload)
+	}
+
+	txDataJSON, err := json.Marshal(txData)
+	if err != nil {
+		return err
+	}
+
+	if err := os.WriteFile(txsFile, txDataJSON, 0644); err != nil {
+		return err
+	}
+
+	log.Printf("Flushed %d transactions to %s", len(w.txBuffer), txsFile)
+
+	return nil
+}

sender/writer_test.go

@@ -0,0 +1,55 @@
+package sender
+
+import (
+	"context"
+	"testing"
+
+	"github.com/sei-protocol/sei-load/config"
+	"github.com/sei-protocol/sei-load/generator"
+	"github.com/sei-protocol/sei-load/generator/scenarios"
+	"github.com/sei-protocol/sei-load/types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTxsWriter_Flush(t *testing.T) {
+	// two evm transfer txs
+	writer := NewTxsWriter(42000, "/tmp")
+
+	loadConfig := &config.LoadConfig{
+		ChainID: 7777,
+	}
+
+	sharedAccounts := types.NewAccountPool(&types.AccountConfig{
+		Accounts:       types.GenerateAccounts(10),
+		NewAccountRate: 0.0,
+	})
+
+	evmScenario := scenarios.CreateScenario(config.Scenario{
+		Name:   "EVMTransfer",
+		Weight: 1,
+	})
+	evmScenario.Deploy(loadConfig, sharedAccounts.NextAccount())
+
+	generator := generator.NewScenarioGenerator(sharedAccounts, evmScenario)
+
+	txs := generator.GenerateN(3)
+
+	writer.Send(context.Background(), txs[0])
+	require.Equal(t, uint64(1), writer.nextHeight)
+	require.Equal(t, uint64(21000), writer.bufferGas)
+	require.Len(t, writer.txBuffer, 1)
+	require.Equal(t, txs[0], writer.txBuffer[0])
+
+	writer.Send(context.Background(), txs[1])
+	require.Equal(t, uint64(1), writer.nextHeight)
+	require.Equal(t, uint64(42000), writer.bufferGas)
+	require.Len(t, writer.txBuffer, 2)
+	require.Equal(t, txs[1], writer.txBuffer[1])
+
+	writer.Send(context.Background(), txs[2])
+	// now should be flushed and have the new tx
+	require.Equal(t, uint64(2), writer.nextHeight)
+	require.Equal(t, uint64(21000), writer.bufferGas)
+	require.Len(t, writer.txBuffer, 1)
+
+}