fix

Author: pompon0

main.go

@@ -160,17 +160,14 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command, args []string) error {
 		}
 
 		// Create shared rate limiter for all workers if TPS is specified
-		var sharedLimiter *rate.Limiter
+		sharedLimiter := rate.NewLimiter(rate.Inf, 1) // no limit by default
 		if settings.TPS > 0 {
 			sharedLimiter = rate.NewLimiter(rate.Limit(settings.TPS), 1)
 			log.Printf("📈 Rate limiting enabled: %.2f TPS shared across all workers", settings.TPS)
-		} else {
-			// No rate limiting
-			sharedLimiter = rate.NewLimiter(rate.Inf, 1)
 		}
 
 		// Create the sender from the config struct
-		snd, err := sender.NewShardedSender(cfg, settings.BufferSize, settings.Workers, sharedLimiter)
+		snd, err := sender.NewShardedSender(cfg, settings.BufferSize, settings.Workers)
 		if err != nil {
 			return fmt.Errorf("failed to create sender: %w", err)
 		}
@@ -240,7 +237,7 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command, args []string) error {
 		}
 
 		// Start the sender (starts all workers)
-		s.SpawnBgNamed("sender", func() error { return snd.Run(ctx) })
+		s.SpawnBgNamed("sender", func() error { return snd.Run(ctx, sharedLimiter) })
 		log.Printf("✅ Connected to %d endpoints", snd.GetNumShards())
 
 		// Perform prewarming if enabled (before starting logger to avoid logging prewarm transactions)

profiles/profiles_test.go

@@ -80,7 +80,7 @@ func TestProfilesAlignment(t *testing.T) {
 			// Use a decoder with DisallowUnknownFields to catch any extra fields
 			decoder := json.NewDecoder(strings.NewReader(string(data)))
 			decoder.DisallowUnknownFields()
-			
+
 			var strictConfig config.LoadConfig
 			if err := decoder.Decode(&strictConfig); err != nil {
 				t.Errorf("Profile %s contains unexpected/unaligned fields: %v", file.Name(), err)

sender/sharded_sender.go

@@ -26,7 +26,7 @@ type ShardedSender struct {
 }
 
 // NewShardedSender creates a new sharded sender with workers for each endpoint
-func NewShardedSender(cfg *config.LoadConfig, bufferSize int, workers int, limiter *rate.Limiter) (*ShardedSender, error) {
+func NewShardedSender(cfg *config.LoadConfig, bufferSize int, workers int) (*ShardedSender, error) {
 	if len(cfg.Endpoints) == 0 {
 		return nil, fmt.Errorf("no endpoints configured")
 	}
@@ -40,7 +40,6 @@ func NewShardedSender(cfg *config.LoadConfig, bufferSize int, workers int, limit
 		workers:    workerList,
 		numShards:  len(cfg.Endpoints),
 		bufferSize: bufferSize,
-		limiter:    limiter,
 	}, nil
 }
 
@@ -51,7 +50,7 @@ func (s *ShardedSender) Run(ctx context.Context, limiter *rate.Limiter) error {
 	s.mu.Unlock()
 	return service.Run(ctx, func(ctx context.Context, s service.Scope) error {
 		for _, worker := range workers {
-			s.Spawn(func() error { return worker.Run(ctx,limiter) })
+			s.Spawn(func() error { return worker.Run(ctx, limiter) })
 		}
 		return nil
 	})

sender/worker.go

@@ -60,7 +60,7 @@ func (w *Worker) Run(ctx context.Context, limiter *rate.Limiter) error {
 	return service.Run(ctx, func(ctx context.Context, s service.Scope) error {
 		// Start multiple worker goroutines that share the same channel
 		for range w.workers {
-			s.Spawn(func() error { return w.processTransactions(ctx,limiter) })
+			s.Spawn(func() error { return w.processTransactions(ctx, limiter) })
 		}
 		return w.watchTransactions(ctx)
 	})
@@ -159,7 +159,7 @@ func (w *Worker) processTransactions(ctx context.Context, limiter *rate.Limiter)
 
 	for ctx.Err() == nil {
 		// Apply rate limiting before getting the next transaction
-		if err := limiter.Wait(ctx); err!=nil {
+		if err := limiter.Wait(ctx); err != nil {
 			return err
 		}
 		tx, err := utils.Recv(ctx, w.txChan)