celestiaorg
diff --git a/‎Makefile‎
Lines changed: 2 additions & 1 deletion b/‎Makefile‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎tools/fibre-txsim/README.md‎
Lines changed: 64 additions & 0 deletions b/‎tools/fibre-txsim/README.md‎
Lines changed: 64 additions & 0 deletions
diff --git a/‎tools/fibre-txsim/main.go‎
Lines changed: 231 additions & 0 deletions b/‎tools/fibre-txsim/main.go‎
Lines changed: 231 additions & 0 deletions
@@ -360,12 +360,13 @@ txsim-build-docker:
 	docker build -t ghcr.io/celestiaorg/txsim -f docker/txsim/Dockerfile  .
 .PHONY: txsim-build-docker
 
-## build-talis-bins: Build celestia-appd, txsim, and latency-monitor binaries for talis VMs (ubuntu 22.04 LTS)
+## build-talis-bins: Build celestia-appd, txsim, latency-monitor, and fibre-txsim binaries for talis VMs (ubuntu 22.04 LTS)
 build-talis-bins:
 	mkdir -p build
 	GOOS=linux GOARCH=amd64 go build -ldflags "$(LDFLAGS_STANDALONE)" -o build/celestia-appd ./cmd/celestia-appd
 	GOOS=linux GOARCH=amd64 go build -o build/txsim ./test/cmd/txsim
 	GOOS=linux GOARCH=amd64 go build -o build/latency-monitor ./tools/latency-monitor 2>/dev/null || echo "latency-monitor not found, skipping"
+	GOOS=linux GOARCH=amd64 go build -o build/fibre-txsim ./tools/fibre-txsim
 .PHONY: build-talis-bins
 
 
 
@@ -0,0 +1,64 @@
+# fibre-txsim
+
+A load-generation tool that submits blobs to a Celestia network through the Fibre protocol. It connects to a validator's gRPC endpoint, creates random blobs, and sends them via `MsgPayForFibre` as fast as possible (or at a configured interval).
+
+This binary is built for Linux and deployed to validator nodes by `make build-talis-bins`. It is started remotely via the `talis fibre-txsim` command.
+
+## Build
+
+```sh
+# Cross-compile for talis VMs (Linux amd64)
+make build-talis-bins
+
+# Build for your local machine (useful for local testing)
+go build -o fibre-txsim ./tools/fibre-txsim/
+```
+
+## Usage
+
+```sh
+fibre-txsim \
+  --chain-id <chain-id> \
+  --grpc-endpoint localhost:9091 \
+  --keyring-dir .celestia-app \
+  --key-name validator \
+  --blob-size 1000000 \
+  --concurrency 4 \
+  --interval 0s
+```
+
+## Flags
+
+| Flag              | Default          | Description                                                                 |
+|-------------------|------------------|-----------------------------------------------------------------------------|
+| `--chain-id`      | *(required)*     | Chain ID of the network                                                     |
+| `--grpc-endpoint` | `localhost:9091` | gRPC endpoint of the validator                                              |
+| `--keyring-dir`   | `.celestia-app`  | Path to the keyring directory                                               |
+| `--key-name`      | `validator`      | Key name in the keyring                                                     |
+| `--blob-size`     | `1000000`        | Size of each blob in bytes                                                  |
+| `--concurrency`   | `1`              | Number of concurrent blob submissions                                       |
+| `--interval`      | `0`              | Delay between blob submissions (`0` = no delay, submit as fast as possible) |
+| `--duration`      | `0`              | How long to run (`0` = until killed with Ctrl+C)                            |
+
+## How it works
+
+1. Connects to a validator via gRPC and initializes a Fibre client.
+2. Spawns up to `--concurrency` goroutines that each:
+   - Generate a random namespace and random blob data of `--blob-size` bytes.
+   - Call `fibreClient.Put()` to submit the blob through the Fibre protocol.
+   - Log the resulting block height, tx hash, and submission latency.
+3. On shutdown (Ctrl+C or `--duration` elapsed), prints a summary with total sent, successes, failures, and average latency.
+
+## Typical deployment
+
+You don't normally run `fibre-txsim` directly. Instead, use `talis fibre-txsim` which SSHes into validators and starts it inside a tmux session:
+
+```sh
+talis fibre-txsim --directory <experiment-dir> \
+  --instances 4 \
+  --concurrency 2 \
+  --blob-size 500000 \
+  --duration 10m
+```
+
+See `tools/talis/fibre.md` for the full experiment workflow.
@@ -0,0 +1,231 @@
+package main
+
+import (
+	"context"
+	"crypto/rand"
+	"flag"
+	"fmt"
+	"math"
+	"os"
+	"os/signal"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/celestiaorg/celestia-app-fibre/v6/app"
+	"github.com/celestiaorg/celestia-app-fibre/v6/app/encoding"
+	"github.com/celestiaorg/celestia-app-fibre/v6/fibre"
+	fibregrpc "github.com/celestiaorg/celestia-app-fibre/v6/fibre/grpc"
+	"github.com/celestiaorg/celestia-app-fibre/v6/pkg/user"
+	valaddrtypes "github.com/celestiaorg/celestia-app-fibre/v6/x/valaddr/types"
+	"github.com/celestiaorg/go-square/v4/share"
+	coregrpc "github.com/cometbft/cometbft/rpc/grpc"
+	"github.com/cosmos/cosmos-sdk/crypto/keyring"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+func main() {
+	var (
+		chainID      string
+		grpcEndpoint string
+		keyringDir   string
+		keyName      string
+		blobSize     int
+		concurrency  int
+		interval     time.Duration
+		duration     time.Duration
+	)
+
+	flag.StringVar(&chainID, "chain-id", "", "chain ID (required)")
+	flag.StringVar(&grpcEndpoint, "grpc-endpoint", "localhost:9091", "gRPC endpoint")
+	flag.StringVar(&keyringDir, "keyring-dir", ".celestia-app", "keyring directory")
+	flag.StringVar(&keyName, "key-name", "validator", "key name in keyring")
+	flag.IntVar(&blobSize, "blob-size", 1000000, "size of each blob in bytes")
+	flag.IntVar(&concurrency, "concurrency", 1, "number of concurrent blob submissions")
+	flag.DurationVar(&interval, "interval", 0, "delay between blob submissions (0 = no delay)")
+	flag.DurationVar(&duration, "duration", 0, "how long to run (0 = until killed)")
+	flag.Parse()
+
+	if chainID == "" {
+		fmt.Fprintln(os.Stderr, "error: --chain-id is required")
+		flag.Usage()
+		os.Exit(1)
+	}
+
+	if err := run(chainID, grpcEndpoint, keyringDir, keyName, blobSize, concurrency, interval, duration); err != nil {
+		fmt.Fprintf(os.Stderr, "error: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+func run(chainID, grpcEndpoint, keyringDir, keyName string, blobSize, concurrency int, interval, duration time.Duration) error {
+	encCfg := encoding.MakeConfig(app.ModuleEncodingRegisters...)
+
+	kr, err := keyring.New(app.Name, keyring.BackendTest, keyringDir, nil, encCfg.Codec)
+	if err != nil {
+		return fmt.Errorf("failed to initialize keyring: %w", err)
+	}
+
+	grpcConn, err := grpc.NewClient(
+		grpcEndpoint,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultCallOptions(
+			grpc.MaxCallSendMsgSize(math.MaxInt32),
+			grpc.MaxCallRecvMsgSize(math.MaxInt32),
+		),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create gRPC connection: %w", err)
+	}
+	defer grpcConn.Close()
+
+	valGet := fibregrpc.NewSetGetter(coregrpc.NewBlockAPIClient(grpcConn))
+	hostReg := fibregrpc.NewHostRegistry(valaddrtypes.NewQueryClient(grpcConn))
+
+	clientCfg := fibre.DefaultClientConfig()
+	clientCfg.ChainID = chainID
+	clientCfg.DefaultKeyName = keyName
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	txClient, err := user.SetupTxClient(ctx, kr, grpcConn, encCfg, user.WithDefaultAccount(keyName))
+	if err != nil {
+		return fmt.Errorf("failed to set up tx client: %w", err)
+	}
+
+	fibreClient, err := fibre.NewClient(txClient, kr, valGet, hostReg, clientCfg)
+	if err != nil {
+		return fmt.Errorf("failed to create fibre client: %w", err)
+	}
+	defer fibreClient.Close()
+
+	// Handle signals
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, os.Interrupt)
+	go func() {
+		<-sigCh
+		fmt.Println("\nReceived interrupt, shutting down...")
+		cancel()
+	}()
+
+	// Apply duration limit if set
+	if duration > 0 {
+		ctx, cancel = context.WithTimeout(ctx, duration)
+		defer cancel()
+	}
+
+	// Stats
+	var (
+		totalSent  atomic.Int64
+		successes  atomic.Int64
+		failures   atomic.Int64
+		totalLatNs atomic.Int64
+	)
+	startTime := time.Now()
+
+	// Semaphore for bounded concurrency
+	sem := make(chan struct{}, concurrency)
+	var wg sync.WaitGroup
+
+	fmt.Println("\nStarting fibre blob spam...")
+
+	// If interval is set, use a ticker to pace blob submissions.
+	// Otherwise, fire as fast as the semaphore allows.
+	var tick <-chan time.Time
+	if interval > 0 {
+		t := time.NewTicker(interval)
+		defer t.Stop()
+		tick = t.C
+	}
+
+	for ctx.Err() == nil {
+		// Wait for the interval tick (if configured)
+		if tick != nil {
+			select {
+			case <-ctx.Done():
+				continue
+			case <-tick:
+			}
+		}
+
+		// Acquire semaphore slot
+		select {
+		case <-ctx.Done():
+			continue
+		case sem <- struct{}{}:
+		}
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			defer func() { <-sem }()
+
+			// Generate random namespace
+			nsID := make([]byte, share.NamespaceVersionZeroIDSize)
+			if _, err := rand.Read(nsID); err != nil {
+				fmt.Printf("error generating namespace: %v\n", err)
+				failures.Add(1)
+				totalSent.Add(1)
+				return
+			}
+			id := make([]byte, 0, share.NamespaceIDSize)
+			id = append(id, share.NamespaceVersionZeroPrefix...)
+			id = append(id, nsID...)
+			ns, err := share.NewNamespace(share.NamespaceVersionZero, id)
+			if err != nil {
+				fmt.Printf("error creating namespace: %v\n", err)
+				failures.Add(1)
+				totalSent.Add(1)
+				return
+			}
+
+			// Generate random blob data
+			data := make([]byte, blobSize)
+			if _, err := rand.Read(data); err != nil {
+				fmt.Printf("error generating blob data: %v\n", err)
+				failures.Add(1)
+				totalSent.Add(1)
+				return
+			}
+
+			t := time.Now()
+			result, err := fibreClient.Put(ctx, ns, data)
+			lat := time.Since(t)
+
+			totalSent.Add(1)
+			if err != nil {
+				if ctx.Err() != nil {
+					return
+				}
+				failures.Add(1)
+				fmt.Printf("error: %v (latency=%s)\n", err, lat)
+				return
+			}
+
+			successes.Add(1)
+			totalLatNs.Add(lat.Nanoseconds())
+			fmt.Printf("height=%d tx=%s latency=%s\n", result.Height, result.TxHash, lat)
+		}()
+	}
+
+	wg.Wait()
+
+	elapsed := time.Since(startTime)
+	s := successes.Load()
+	f := failures.Load()
+	var avgLat time.Duration
+	if s > 0 {
+		avgLat = time.Duration(totalLatNs.Load() / s)
+	}
+
+	fmt.Printf("\n--- Summary ---\n")
+	fmt.Printf("Duration:   %s\n", elapsed.Truncate(time.Second))
+	fmt.Printf("Total sent: %d\n", totalSent.Load())
+	fmt.Printf("Successes:  %d\n", s)
+	fmt.Printf("Failures:   %d\n", f)
+	fmt.Printf("Avg latency (success): %s\n", avgLat)
+
+	return nil
+}