M10(1/5): loadgen pacer + --ramp-up + --cluster-id + 3-instance compose profile

W1: loadgen scale-out for M10 (5k/s via 3 × 1.7k/s instances)

New:
- loadgen/internal/pacer/pacer.go — linear ramp pacer; replaces fixed ticker
  in http driver. Supports ramp-up window, injectable clock for tests.
- loadgen/internal/pacer/pacer_test.go — 4 tests, all PASS
- loadgen/cmd/http/main.go — uses pacer, adds --ramp-up + --cluster-id + --instance
- loadgen/cmd/mqtt/main.go — adds --ramp-up + --cluster-id + --instance flags
- loadgen/cmd/ws/main.go — adds --ramp-up + --cluster-id + --instance + --metrics flags
- docker-compose.yml — adds loadgen-http-{1,2,3} under profile [loadgen-m10]
  (3 × acme-00X sources, same prom-prod company, per-company cap 10k/s > 5.1k/s total)
- deploy/prometheus/prometheus.yml — adds loadgen-m10 scrape job for the 3 instances
  (loadgen-http-1:8891, loadgen-http-2:8891, loadgen-http-3:8891)

Driver behavior: http now uses the pacer for rate control (supports ramp-up).
mqtt/ws: flags added for consistency; ramp-up not yet wired (count-based drivers
don't need duration-based pacing for M10 smoke purposes).

deploy/prometheus/prometheus.yml

@@ -38,6 +38,11 @@ scrape_configs:
     static_configs:
       - targets: ['loadgen-http:8891']
 
+  # M10: 3-instance cluster (loadgen-m10 profile)
+  - job_name: loadgen-m10
+    static_configs:
+      - targets: ['loadgen-http-1:8891', 'loadgen-http-2:8891', 'loadgen-http-3:8891']
+
   # ── Prometheus self-monitoring ────────────────────────────
   - job_name: prometheus
     static_configs:

docker-compose.yml

@@ -321,10 +321,66 @@ services:
       - --rate=50
       - --duration=30s
       - --metrics=:8891
+      - --instance=loadgen-http
+      - --cluster-id=smoke
     profiles: ["loadgen"]
     depends_on:
       ingestd: { condition: service_started }
 
+  # M10: 3-instance cluster hitting 5k/s. Each instance does ~1.7k/s.
+  # Per-company cap is 10k/s, so 3 × 1.7k = 5.1k is safe.
+  # Use `docker compose --profile loadgen-m10 up -d` to bring up all 3.
+  loadgen-http-1:
+    build: .
+    command:
+      - /app/loadgen-http
+      - --target=http://ingestd:8800
+      - --api-key=acme-001:prom-prod:s3cret-acme-001
+      - --mode=normal
+      - --rate=1700
+      - --duration=10m
+      - --ramp-up=30s
+      - --metrics=:8891
+      - --instance=loadgen-http-1
+      - --cluster-id=m10
+    profiles: ["loadgen-m10"]
+    depends_on:
+      ingestd: { condition: service_started }
+
+  loadgen-http-2:
+    build: .
+    command:
+      - /app/loadgen-http
+      - --target=http://ingestd:8800
+      - --api-key=acme-002:prom-prod:s3cret-acme-002
+      - --mode=normal
+      - --rate=1700
+      - --duration=10m
+      - --ramp-up=30s
+      - --metrics=:8891
+      - --instance=loadgen-http-2
+      - --cluster-id=m10
+    profiles: ["loadgen-m10"]
+    depends_on:
+      ingestd: { condition: service_started }
+
+  loadgen-http-3:
+    build: .
+    command:
+      - /app/loadgen-http
+      - --target=http://ingestd:8800
+      - --api-key=acme-003:prom-prod:s3cret-acme-003
+      - --mode=normal
+      - --rate=1700
+      - --duration=10m
+      - --ramp-up=30s
+      - --metrics=:8891
+      - --instance=loadgen-http-3
+      - --cluster-id=m10
+    profiles: ["loadgen-m10"]
+    depends_on:
+      ingestd: { condition: service_started }
+
 volumes:
   pgdata: {}
   natsdata: {}

loadgen/cmd/http/main.go

@@ -2,11 +2,17 @@
 // --mode normal only for now; other profiles land as the protocols
 // come online.
 //
-// Example:
+// Example (single instance):
 //
 //	loadgen-http --target http://localhost:8800 \
 //	  --api-key acme-001:prom-prod:s3cret \
-//	  --mode normal --rate 100 --duration 30s
+//	  --mode normal --rate 100 --duration 30s --ramp-up 10s
+//
+// Example (3-instance cluster hitting 5k/s):
+//
+//	loadgen-http-1 --rate 1700 --duration 10m --ramp-up 30s --cluster-id m10-run
+//	loadgen-http-2 --rate 1700 --duration 10m --ramp-up 30s --cluster-id m10-run
+//	loadgen-http-3 --rate 1700 --duration 10m --ramp-up 30s --cluster-id m10-run
 package main
 
 import (
@@ -32,6 +38,7 @@ import (
 	"time"
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
+	"git3.techno-world.net/lrosales/broad-announce/loadgen/internal/pacer"
 )
 
 func main() {
@@ -39,12 +46,15 @@ func main() {
 		target     = flag.String("target", "http://localhost:8800", "ingestd base URL")
 		apiKey     = flag.String("api-key", "", "company_id:source_id:secret")
 		mode       = flag.String("mode", "normal", "profile: normal|burst|stress")
-		rate       = flag.Int("rate", 100, "target alerts/sec per instance")
+		rate       = flag.Int("rate", 100, "target alerts/sec (after ramp-up)")
 		duration   = flag.Duration("duration", 30*time.Second, "total run time")
+		rampUp     = flag.Duration("ramp-up", 0, "linear ramp from 0 to --rate over this duration (default 0 = instant)")
 		conc       = flag.Int("concurrency", 16, "concurrent HTTP requests in flight")
 		metrics    = flag.String("metrics", ":8891", "Prometheus metrics listen addr (empty to disable)")
+		clusterID  = flag.String("cluster-id", "default", "tag added as label on all loadgen metrics")
+		instance   = flag.String("instance", "default", "loadgen instance name (e.g. loadgen-http-1)")
 		dedupePct  = flag.Int("dedupe-pct", 30, "percent of alerts sharing a dedupe_key (normal mode)")
-		dedupeKey  = flag.String("dedupe-key", "", "force a specific dedupe_key on every alert (overrides --dedupe-pct; useful for M6 ×N smoke tests)")
+		dedupeKey  = flag.String("dedupe-key", "", "force a specific dedupe_key on every alert (overrides --dedupe-pct)")
 		companyN   = flag.Int("companies", 1, "how many fake companies to cycle through")
 		payloadB   = flag.Int("payload-bytes", 256, "approximate payload size in bytes (Data map)")
 	)
@@ -59,22 +69,26 @@ func main() {
 		os.Exit(2)
 	}
 	company, source, secret := parts[0], parts[1], parts[2]
-	_ = company
-	_ = source
-	_ = secret
 
 	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
+	logger.Info("starting",
+		"target", *target,
+		"rate", *rate,
+		"ramp_up", *rampUp,
+		"duration", *duration,
+		"concurrency", *conc,
+		"cluster_id", *clusterID,
+		"instance", *instance,
+	)
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
 
-	limiter := newRateLimiter(*rate)
-
 	var (
-		sent     atomic.Uint64
-		failed   atomic.Uint64
-		dupes    atomic.Uint64
-		rlHits   atomic.Uint64
+		sent   atomic.Uint64
+		failed atomic.Uint64
+		dupes  atomic.Uint64
+		rlHits atomic.Uint64
 	)
 
 	// Spawn workers.
@@ -89,9 +103,6 @@ func main() {
 				if ctx.Err() != nil {
 					return
 				}
-				if err := limiter.Wait(ctx); err != nil {
-					return
-				}
 				a := mkAlert(*mode, *companyN, *dedupePct, *dedupeKey, *payloadB)
 				if err := sendOne(ctx, client, *target, company, source, []byte(secret), a); err != nil {
 					failed.Add(1)
@@ -109,18 +120,21 @@ func main() {
 		}(i)
 	}
 
-	// Producer.
+	// Producer with pacer (supports linear ramp-up).
 	producerCtx, cancelProducer := context.WithTimeout(ctx, *duration)
 	defer cancelProducer()
+
+	p := pacer.New(*rate, *rampUp)
+	tickCh, stopPacer := p.Tick(producerCtx)
+	defer stopPacer()
+
 	go func() {
-		ticker := time.NewTicker(time.Second / time.Duration(*rate+1))
-		defer ticker.Stop()
 		for {
 			select {
 			case <-producerCtx.Done():
 				close(jobs)
 				return
-			case <-ticker.C:
+			case <-tickCh:
 				select {
 				case jobs <- struct{}{}:
 				default:
@@ -132,7 +146,7 @@ func main() {
 
 	// Metrics endpoint.
 	if *metrics != "" {
-		go runMetrics(*metrics, &sent, &failed, &dupes, &rlHits)
+		go runMetrics(*metrics, *clusterID, *instance, *mode, &sent, &failed, &dupes, &rlHits)
 	}
 
 	// Status ticker.
@@ -164,17 +178,10 @@ func main() {
 }
 
 func mkAlert(mode string, companies, dedupePct int, dedupeKey string, payloadB int) alert.Alert {
-	// Cycle through a few company_ids; the source is fixed for the
-	// run (loadgen is one source).
 	companyID := fmt.Sprintf("acme-%03d", (rand.IntN(companies) + 1))
 	severity := pickSeverity(mode)
 	category := pickCategory(severity)
 
-	// Dedupe: dedupe_pct % of alerts share a fixed "burst" key
-	// within a minute. For the simpler "normal" mode, we just
-	// include a stable dedupe key on a fraction of alerts.
-	// M6: --dedupe-key forces a specific key on every alert,
-	// which is what the ×N smoke test needs (one key, N copies).
 	dk := ""
 	if dedupeKey != "" {
 		dk = dedupeKey
@@ -182,7 +189,6 @@ func mkAlert(mode string, companies, dedupePct int, dedupeKey string, payloadB i
 		dk = fmt.Sprintf("burst:%s:probe", category)
 	}
 
-	// Synthetic data roughly `payloadB` bytes.
 	data := map[string]string{
 		"host":    fmt.Sprintf("host-%d", rand.IntN(100)),
 		"probe":   category,
@@ -202,7 +208,6 @@ func mkAlert(mode string, companies, dedupePct int, dedupeKey string, payloadB i
 }
 
 func pickSeverity(mode string) alert.Severity {
-	// 70/25/4/1 split for normal, even for stress
 	r := rand.IntN(100)
 	switch {
 	case r < 70:
@@ -278,49 +283,19 @@ func isRateLimited(err error) bool {
 	return ok
 }
 
-// rateLimiter is a simple per-second pacer. Not a token bucket; for
-// loadgen we want "send at most N per second" averaged, not bursty.
-type rateLimiter struct {
-	perSec int
-	last   time.Time
-	mu     sync.Mutex
-}
-
-func newRateLimiter(perSec int) *rateLimiter { return &rateLimiter{perSec: perSec} }
-
-func (r *rateLimiter) Wait(ctx context.Context) error {
-	if r.perSec <= 0 {
-		return nil
-	}
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	now := time.Now()
-	interval := time.Second / time.Duration(r.perSec)
-	if d := interval - now.Sub(r.last); d > 0 {
-		t := time.NewTimer(d)
-		defer t.Stop()
-		r.mu.Unlock()
-		select {
-		case <-ctx.Done():
-			r.mu.Lock()
-			return ctx.Err()
-		case <-t.C:
-		}
-		r.mu.Lock()
-	}
-	r.last = time.Now()
-	return nil
-}
-
-func runMetrics(addr string, sent, failed, dupes, rlHits *atomic.Uint64) {
+func runMetrics(addr, clusterID, instance, mode string, sent, failed, dupes, rlHits *atomic.Uint64) {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
 		fmt.Fprintf(w, "# HELP loadgen_alerts_sent_total Alerts successfully accepted.\n")
 		fmt.Fprintf(w, "# TYPE loadgen_alerts_sent_total counter\n")
-		fmt.Fprintf(w, "loadgen_alerts_sent_total %d\n", sent.Load())
-		fmt.Fprintf(w, "loadgen_alerts_failed_total %d\n", failed.Load())
-		fmt.Fprintf(w, "loadgen_dedupe_hits_total %d\n", dupes.Load())
-		fmt.Fprintf(w, "loadgen_rate_limited_total %d\n", rlHits.Load())
+		fmt.Fprintf(w, "loadgen_alerts_sent_total{profile=%q,instance=%q,cluster_id=%q,mode=%q} %d\n",
+			mode, instance, clusterID, mode, sent.Load())
+		fmt.Fprintf(w, "loadgen_alerts_failed_total{reason=%q,instance=%q,cluster_id=%q} %d\n",
+			"send", instance, clusterID, failed.Load())
+		fmt.Fprintf(w, "loadgen_dedupe_hits_total{instance=%q,cluster_id=%q} %d\n",
+			instance, clusterID, dupes.Load())
+		fmt.Fprintf(w, "loadgen_rate_limited_total{instance=%q,cluster_id=%q} %d\n",
+			instance, clusterID, rlHits.Load())
 	})
 	srv := &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
 	_ = srv.ListenAndServe()
@@ -331,4 +306,4 @@ func max(a, b int) int {
 		return a
 	}
 	return b
-}
+}

loadgen/cmd/mqtt/main.go

@@ -28,6 +28,7 @@ import (
 	"fmt"
 	"log/slog"
 	"math/rand/v2"
+	"net/http"
 	"os"
 	"os/signal"
 	"strings"
@@ -45,11 +46,14 @@ func main() {
 		apiKey    = flag.String("api-key", "", "company:source:secret")
 		count     = flag.Int("count", 10, "total alerts to send")
 		rate      = flag.Int("rate", 10, "target alerts/sec")
+		rampUp    = flag.Duration("ramp-up", 0, "linear ramp from 0 to --rate over this duration (default 0 = instant)")
 		mode      = flag.String("mode", "normal", "profile: normal|burst")
 		dedupePct = flag.Int("dedupe-pct", 30, "percent sharing a dedupe_key (normal)")
-		dedupeKey = flag.String("dedupe-key", "", "force a specific dedupe_key on every alert (overrides --dedupe-pct; useful for M6 ×N smoke tests)")
+		dedupeKey = flag.String("dedupe-key", "", "force a specific dedupe_key on every alert")
 		metrics   = flag.String("metrics", "", "Prometheus metrics listen addr (empty to disable)")
 		timeout   = flag.Duration("duration", 30*time.Second, "max run time")
+		clusterID = flag.String("cluster-id", "default", "tag added as label on all loadgen metrics")
+		instance  = flag.String("instance", "default", "loadgen instance name (e.g. loadgen-mqtt-1)")
 	)
 	flag.Parse()
 	if *apiKey == "" {
@@ -65,6 +69,15 @@ func main() {
 	username := fmt.Sprintf("%s-%s", source, company) // matches auth csv
 
 	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
+	logger.Info("starting",
+		"broker", *broker,
+		"rate", *rate,
+		"ramp_up", *rampUp,
+		"count", *count,
+		"duration", *timeout,
+		"cluster_id", *clusterID,
+		"instance", *instance,
+	)
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
 
@@ -151,7 +164,10 @@ func main() {
 	if failed.Load() > 0 {
 		os.Exit(1)
 	}
-	_ = metrics // reserved for M9
+	// Metrics endpoint (registered but currently a no-op; wired in M10)
+	if *metrics != "" {
+		go runMQTTMetrics(*metrics, *clusterID, *instance, &sent, &failed, &dupes)
+	}
 }
 
 // makeAlert generates one alert. Same shape as loadgen/cmd/http.
@@ -210,3 +226,19 @@ func isUnique(dedupePct, i int) *bool {
 	b := i == 0 || dedupePct == 0 || rand.IntN(100) >= dedupePct
 	return &b
 }
+
+func runMQTTMetrics(addr, clusterID, instance string, sent, failed, dupes *atomic.Uint64) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, "# HELP loadgen_alerts_sent_total Alerts successfully accepted.\n")
+		fmt.Fprintf(w, "# TYPE loadgen_alerts_sent_total counter\n")
+		fmt.Fprintf(w, "loadgen_alerts_sent_total{instance=%q,cluster_id=%q} %d\n",
+			instance, clusterID, sent.Load())
+		fmt.Fprintf(w, "loadgen_alerts_failed_total{instance=%q,cluster_id=%q} %d\n",
+			instance, clusterID, failed.Load())
+		fmt.Fprintf(w, "loadgen_dedupe_hits_total{instance=%q,cluster_id=%q} %d\n",
+			instance, clusterID, dupes.Load())
+	})
+	srv := &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
+	_ = srv.ListenAndServe()
+}

loadgen/cmd/ws/main.go

@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"log/slog"
 	"math/rand/v2"
+	"net/http"
 	"os"
 	"os/signal"
 	"strings"
@@ -41,11 +42,15 @@ func main() {
 		apiKey    = flag.String("api-key", "", "company:source:secret")
 		count     = flag.Int("count", 10, "total alerts to send")
 		rate      = flag.Int("rate", 10, "target alerts/sec")
+		rampUp    = flag.Duration("ramp-up", 0, "linear ramp from 0 to --rate over this duration (default 0 = instant)")
 		mode      = flag.String("mode", "normal", "profile: normal|burst")
 		dedupePct = flag.Int("dedupe-pct", 30, "percent sharing a dedupe_key (normal)")
 		dedupeKey = flag.String("dedupe-key", "", "force a specific dedupe_key on every alert (M6 ×N smoke test)")
 		timeout   = flag.Duration("duration", 30*time.Second, "max run time")
 		concurrencyFlag = flag.Int("concurrency", 1, "parallel WS connections (each one is one source)")
+		clusterID = flag.String("cluster-id", "default", "tag added as label on all loadgen metrics")
+		instance  = flag.String("instance", "default", "loadgen instance name (e.g. loadgen-ws-1)")
+		metricsAddr = flag.String("metrics", "", "Prometheus metrics listen addr (empty to disable)")
 	)
 	flag.Parse()
 	if *apiKey == "" {
@@ -60,6 +65,15 @@ func main() {
 	company, source, secret := parts[0], parts[1], parts[2]
 
 	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
+	logger.Info("starting",
+		"target", *target,
+		"rate", *rate,
+		"ramp_up", *rampUp,
+		"count", *count,
+		"concurrency", *concurrencyFlag,
+		"cluster_id", *clusterID,
+		"instance", *instance,
+	)
 	_, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
 
@@ -143,6 +157,9 @@ func main() {
 	if failed.Load() > 0 {
 		os.Exit(1)
 	}
+	if *metricsAddr != "" {
+		go runWSMetrics(*metricsAddr, *clusterID, *instance, &sent, &failed, &dupes)
+	}
 }
 
 // makeAlert is the same shape as loadgen/cmd/http and
@@ -210,3 +227,19 @@ func isUnique(dedupePct, idx int) bool {
 	}
 	return rand.IntN(100) >= dedupePct
 }
+
+func runWSMetrics(addr, clusterID, instance string, sent, failed, dupes *atomic.Uint64) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, "# HELP loadgen_alerts_sent_total Alerts successfully accepted.\n")
+		fmt.Fprintf(w, "# TYPE loadgen_alerts_sent_total counter\n")
+		fmt.Fprintf(w, "loadgen_alerts_sent_total{instance=%q,cluster_id=%q} %d\n",
+			instance, clusterID, sent.Load())
+		fmt.Fprintf(w, "loadgen_alerts_failed_total{instance=%q,cluster_id=%q} %d\n",
+			instance, clusterID, failed.Load())
+		fmt.Fprintf(w, "loadgen_dedupe_hits_total{instance=%q,cluster_id=%q} %d\n",
+			instance, clusterID, dupes.Load())
+	})
+	srv := &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
+	_ = srv.ListenAndServe()
+}

loadgen/internal/pacer/pacer.go

@@ -0,0 +1,117 @@
+package pacer
+
+import (
+	"context"
+	"time"
+)
+
+// Pacer produces a tick channel that controls send rate.
+// It supports an optional linear ramp from 0 to target rate
+// over a configurable warm-up window.
+type Pacer struct {
+	// Target alerts per second after ramp-up completes.
+	TargetPerSec int
+	// Ramp-up duration. 0 = instant (use full target rate immediately).
+	RampDuration time.Duration
+	// Clock is injectable for testing.
+	Clock func() time.Time
+}
+
+// New returns a Pacer configured for targetPerSec sustained rate
+// with an optional rampDuration warm-up window.
+func New(targetPerSec int, rampDuration time.Duration) *Pacer {
+	return &Pacer{
+		TargetPerSec: targetPerSec,
+		RampDuration: rampDuration,
+		Clock:        time.Now,
+	}
+}
+
+// Tick returns a channel that fires at the appropriate interval
+// for the current phase (ramp or steady-state).
+// The returned stop function blocks until the pacer goroutine exits.
+// Callers must drain the returned channel to avoid blocking the
+// pacer's internal goroutine.
+func (p *Pacer) Tick(ctx context.Context) (<-chan struct{}, func()) {
+	stopped := make(chan struct{})
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+		now := p.Clock()
+		rampStart := now
+
+		if p.TargetPerSec <= 0 || p.RampDuration == 0 {
+			// Steady-state only: fixed interval.
+			ticker := time.NewTicker(intervalFor(p.TargetPerSec))
+			defer ticker.Stop()
+			for {
+				select {
+				case <-ctx.Done():
+					close(stopped)
+					return
+				case <-ticker.C:
+					select {
+					case stopped <- struct{}{}:
+					default:
+					}
+				}
+			}
+		}
+
+		// Ramp-up phase: rate increases linearly from 0 to TargetPerSec.
+		// We recalculate the next tick interval after each tick.
+		for {
+			elapsed := p.Clock().Sub(rampStart)
+			if elapsed >= p.RampDuration {
+				break // fall through to steady-state
+			}
+			// Linear interpolation: fraction of ramp completed.
+			fraction := float64(elapsed) / float64(p.RampDuration)
+			currentRate := int(float64(p.TargetPerSec) * fraction)
+			if currentRate <= 0 {
+				currentRate = 1
+			}
+			tickAfter := intervalFor(currentRate)
+			select {
+			case <-ctx.Done():
+				close(stopped)
+				return
+			case <-time.After(tickAfter):
+				select {
+				case stopped <- struct{}{}:
+				default:
+				}
+			}
+		}
+
+		// Steady-state: fixed interval at full target rate.
+		steady := time.NewTicker(intervalFor(p.TargetPerSec))
+		defer steady.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				close(stopped)
+				return
+			case <-steady.C:
+				select {
+				case stopped <- struct{}{}:
+				default:
+				}
+			}
+		}
+	}()
+
+	return stopped, func() {
+		<-done
+	}
+}
+
+// intervalFor returns the inter-tick interval for a given rate.
+// rate=0 returns a very slow ticker (will never fire in practice).
+func intervalFor(rate int) time.Duration {
+	if rate <= 0 {
+		return time.Hour // effectively stopped
+	}
+	return time.Second / time.Duration(rate)
+}

loadgen/internal/pacer/pacer_test.go

@@ -0,0 +1,98 @@
+package pacer
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestPacer_Instant(t *testing.T) {
+	p := New(100, 0) // 100/s, no ramp
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+
+	ch, stop := p.Tick(ctx)
+	defer stop()
+
+	var count int
+	for range ch {
+		count++
+		if count > 25 {
+			// Should fire ~20 times in 200ms at 100/s
+			break
+		}
+	}
+	if count < 15 {
+		t.Errorf("instant pacer: expected ≥15 ticks in 200ms, got %d", count)
+	}
+}
+
+func TestPacer_RampLinear(t *testing.T) {
+	// This test uses a mock clock that advances with each call.
+	// Total wall clock needed: 100ms (ramp) + 200ms (steady) = 300ms.
+	p := New(100, 100*time.Millisecond) // 100/s, 100ms ramp
+	clockMs := 0
+	p.Clock = func() time.Time {
+		clockMs += 10 // each p.Clock call advances 10ms
+		return time.Date(2000, 1, 1, 0, 0, 0, clockMs*1_000_000, time.UTC)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	ch, stop := p.Tick(ctx)
+	defer stop()
+
+	var count int
+	for range ch {
+		count++
+		if count > 50 {
+			break
+		}
+	}
+	// With mock clock: ramp phase (100ms) + steady phase (~200ms of ticks)
+	// At 100/s, 1 tick per 10ms. In 300ms mock time, should get ~30 ticks.
+	if count < 20 {
+		t.Errorf("ramp pacer: expected ≥20 ticks in 300ms mock time, got %d", count)
+	}
+}
+
+func TestPacer_Overshoot(t *testing.T) {
+	// rate=0 should return a ticker that essentially never fires.
+	p := New(0, 0)
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+
+	ch, stop := p.Tick(ctx)
+	defer stop()
+
+	select {
+	case _, ok := <-ch:
+		if ok {
+			t.Error("channel should be closed for rate=0")
+		}
+	case <-time.After(60 * time.Millisecond):
+		// expected — no ticks fired
+	}
+}
+
+func TestPacer_ContextCancel(t *testing.T) {
+	p := New(10_000, 0) // fast rate
+	ctx, cancel := context.WithCancel(context.Background())
+
+	ch, stop := p.Tick(ctx)
+
+	// Cancel immediately
+	cancel()
+	stop()
+
+	// Channel should close (no tick sent after cancel)
+	select {
+	case _, ok := <-ch:
+		if ok {
+			t.Error("channel should be closed after context cancel")
+		}
+	case <-time.After(50 * time.Millisecond):
+		// ok — closed within 50ms
+	}
+}