Merge pull request #24 from somatic-labs/faddat/grpc2

use 10 positions

.cursorignore

@@ -0,0 +1,3 @@
+# Add directories or file patterns to ignore during indexing (e.g. foo/ or *.csv)
+emails/
+configurations/

.cursorrules

@@ -0,0 +1,23 @@
+// Hardhat .cursorrules
+
+// HTMX and Go best practices
+const htmxGoBestPractices = [
+"follow practices suited for cosmos-sdk v0.50.x",
+"follow practices suited for ibc-go v8.x.x"
+];
+
+// Folder structure
+const folderStructure = `
+  main.go
+lib/
+
+modules/
+go.mod
+go.sum
+`;
+
+// Additional instructions
+const additionalInstructions = `
+1. We are building a high speed test framework for mainnets
+2. We should aim for this application to be precise and performant
+`;

.github/workflows/clippy.yml

@@ -0,0 +1,29 @@
+# .github/workflows/clippy.yml
+name: Clippy
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+  workflow_dispatch:
+
+jobs:
+  clippy:
+    name: Run Clippy
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: stable
+          override: true
+          components: clippy
+      - name: Run Clippy
+        uses: actions-rs/clippy-check@v1
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+          args: --all-features --manifest-path contracts/statefilestore/Cargo.toml -- -D warnings

.github/workflows/test.yml

@@ -0,0 +1,21 @@
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+name: Test
+jobs:
+  test:
+    strategy:
+      matrix:
+        go-version: [1.23.x]
+        os: [ubuntu-latest, macos-latest]
+    runs-on: ${{ matrix.os }}
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version: ${{ matrix.go-version }}
+      - run: go test ./...

.gitignore

@@ -1,2 +1,3 @@
+target/
 **seedphrase
-**meteorite
+**meteorite

README.md

@@ -40,7 +40,14 @@ This will initiate the testing suite with your specified configurations.
 
 ## Important Notes
 
-- **Responsible Use**: meteorite is designed for use on test networks and should be used responsibly. Ensure you have proper authorization before testing on any network you do not own or operate.
+- **Responsible Use**: meteorite is designed for use on mainnets.  The tokens you own grant you the right to make any valid transaction.  Like any user of any chain, meteorite can only make valid transactions.  
+  * Test your favorite mainnet, yourself, and make sure that you can't get Luna'd
+  * Test your favorite mainnet, yourself, and make sure that you can't get Levana'd
+
+Over $70,001,400,000 has been lost to the class of issues named p2p storms.  
+
+If you're investing, test the chain.  It's free if you tweet about using meteorite.
+
 - **Valid Transactions Only**: The tool operates within the bounds of valid transactions explicitly supported by the chains it tests.
 - **Reporting Issues**: For questions about meteorite's capabilities or to report potential security issues, please contact the project maintainers through the appropriate channels listed in this repository.
 

broadcast/broadcast.go

@@ -1,20 +1,58 @@
 package broadcast
 
 import (
-	"context"
 	"fmt"
-	"log"
+	"time"
 
-	cometrpc "github.com/cometbft/cometbft/rpc/client/http"
 	coretypes "github.com/cometbft/cometbft/rpc/core/types"
-	tmtypes "github.com/cometbft/cometbft/types"
 	transfertypes "github.com/cosmos/ibc-go/v8/modules/apps/transfer/types"
+	"github.com/somatic-labs/meteorite/lib"
+	types "github.com/somatic-labs/meteorite/types"
 
 	"github.com/cosmos/cosmos-sdk/codec"
 	codectypes "github.com/cosmos/cosmos-sdk/codec/types"
 	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
 )
 
+// Add these at the top of the file
+type TimingMetrics struct {
+	PrepStart  time.Time
+	SignStart  time.Time
+	BroadStart time.Time
+	Complete   time.Time
+	Position   int
+}
+
+func (t *TimingMetrics) LogTiming(sequence uint64, success bool, err error) {
+	prepTime := t.SignStart.Sub(t.PrepStart)
+	signTime := t.BroadStart.Sub(t.SignStart)
+	broadcastTime := t.Complete.Sub(t.BroadStart)
+	totalTime := t.Complete.Sub(t.PrepStart)
+
+	status := "SUCCESS"
+	if !success {
+		status = "FAILED"
+	}
+
+	fmt.Printf("[POS-%d] %s Transaction %s: seq=%d prep=%v sign=%v broadcast=%v total=%v%s\n",
+		t.Position,
+		time.Now().Format("15:04:05.000"),
+		status,
+		sequence,
+		prepTime,
+		signTime,
+		broadcastTime,
+		totalTime,
+		formatError(err))
+}
+
+func formatError(err error) string {
+	if err != nil {
+		return fmt.Sprintf(" error=\"%v\"", err)
+	}
+	return ""
+}
+
 var cdc = codec.NewProtoCodec(codectypes.NewInterfaceRegistry())
 
 func init() {
@@ -24,24 +62,88 @@ func init() {
 
 // Transaction broadcasts the transaction bytes to the given RPC endpoint.
 func Transaction(txBytes []byte, rpcEndpoint string) (*coretypes.ResultBroadcastTx, error) {
-	cmtCli, err := cometrpc.New(rpcEndpoint, "/websocket")
+	client, err := GetClient(rpcEndpoint)
 	if err != nil {
-		log.Fatal(err)
+		return nil, err
 	}
 
-	t := tmtypes.Tx(txBytes)
+	return client.Transaction(txBytes)
+}
 
-	ctx := context.Background()
-	res, err := cmtCli.BroadcastTxSync(ctx, t)
-	if err != nil {
-		fmt.Println("Error at broadcast:", err)
-		return nil, err
+// Loop handles the main transaction broadcasting logic
+func Loop(
+	txParams types.TransactionParams,
+	batchSize int,
+	position int,
+) (successfulTxns, failedTxns int, responseCodes map[uint32]int, updatedSequence uint64) {
+	successfulTxns = 0
+	failedTxns = 0
+	responseCodes = make(map[uint32]int)
+	sequence := txParams.Sequence
+
+	for i := 0; i < batchSize; i++ {
+		currentSequence := sequence
+		metrics := &TimingMetrics{
+			PrepStart: time.Now(),
+			Position:  position,
+		}
+
+		metrics.SignStart = time.Now()
+		metrics.BroadStart = time.Now()
+		resp, _, err := SendTransactionViaRPC(txParams, currentSequence)
+		metrics.Complete = time.Now()
+
+		if err != nil {
+			metrics.LogTiming(currentSequence, false, err)
+			failedTxns++
+
+			if resp != nil && resp.Code == 32 {
+				newSeq, success, newResp := handleSequenceMismatch(txParams, position, sequence, err)
+				sequence = newSeq
+				if success {
+					successfulTxns++
+					responseCodes[newResp.Code]++
+				}
+				continue
+			}
+			continue
+		}
+
+		metrics.LogTiming(currentSequence, true, nil)
+		successfulTxns++
+		responseCodes[resp.Code]++
+		sequence++
+	}
+
+	updatedSequence = sequence
+	return successfulTxns, failedTxns, responseCodes, updatedSequence
+}
+
+// handleSequenceMismatch handles the case where a transaction fails due to sequence mismatch
+func handleSequenceMismatch(txParams types.TransactionParams, position int, sequence uint64, err error) (uint64, bool, *coretypes.ResultBroadcastTx) {
+	expectedSeq, parseErr := lib.ExtractExpectedSequence(err.Error())
+	if parseErr != nil {
+		fmt.Printf("[POS-%d] Failed to parse expected sequence: %v\n", position, parseErr)
+		return sequence, false, nil
 	}
 
-	if res.Code != 0 {
-		// Return an error containing the code and log message
-		return res, fmt.Errorf("broadcast error code %d: %s", res.Code, res.Log)
+	fmt.Printf("[POS-%d] Set sequence to expected value %d due to mismatch\n", position, expectedSeq)
+
+	metrics := &TimingMetrics{
+		PrepStart: time.Now(),
+		Position:  position,
+	}
+
+	metrics.SignStart = time.Now()
+	metrics.BroadStart = time.Now()
+	resp, _, err := SendTransactionViaRPC(txParams, expectedSeq)
+	metrics.Complete = time.Now()
+
+	if err != nil {
+		metrics.LogTiming(expectedSeq, false, err)
+		return expectedSeq, false, nil
 	}
 
-	return res, nil
+	metrics.LogTiming(expectedSeq, true, nil)
+	return expectedSeq + 1, true, resp
 }

broadcast/client.go

@@ -0,0 +1,68 @@
+package broadcast
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	cometrpc "github.com/cometbft/cometbft/rpc/client/http"
+	coretypes "github.com/cometbft/cometbft/rpc/core/types"
+	tmtypes "github.com/cometbft/cometbft/types"
+)
+
+type Client struct {
+	client *cometrpc.HTTP
+}
+
+var (
+	clients    = make(map[string]*Client)
+	clientsMux sync.RWMutex
+)
+
+func GetClient(rpcEndpoint string) (*Client, error) {
+	clientsMux.RLock()
+	if client, exists := clients[rpcEndpoint]; exists {
+		clientsMux.RUnlock()
+		return client, nil
+	}
+	clientsMux.RUnlock()
+
+	// If client doesn't exist, acquire write lock and create it
+	clientsMux.Lock()
+	defer clientsMux.Unlock()
+
+	// Double-check after acquiring write lock
+	if client, exists := clients[rpcEndpoint]; exists {
+		return client, nil
+	}
+
+	// Create new client
+	cmtCli, err := cometrpc.New(rpcEndpoint, "/websocket")
+	if err != nil {
+		return nil, err
+	}
+
+	client := &Client{
+		client: cmtCli,
+	}
+	clients[rpcEndpoint] = client
+	return client, nil
+}
+
+func (b *Client) Transaction(txBytes []byte) (*coretypes.ResultBroadcastTx, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	t := tmtypes.Tx(txBytes)
+	res, err := b.client.BroadcastTxSync(ctx, t)
+	if err != nil {
+		return nil, err
+	}
+
+	if res.Code != 0 {
+		return res, fmt.Errorf("broadcast error code %d: %s", res.Code, res.Log)
+	}
+
+	return res, nil
+}