NATS latency ramping up

repro_latency_test.go

package server

import (
	"context"
	"fmt"
	"math"
	"sort"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// TestReproLatencyRamp reproduces a latency regression observed since v2.12.1.
// It measures the time between a JetStream message's server-side timestamp
// (Metadata.Timestamp) and when the ordered pull consumer processes it.
// On affected versions the p50/p95/p99 ramp up over time; on v2.12.0 they stay flat.
//
// Run with a long timeout to observe the ramp:
//
//	go test -run TestReproLatencyRamp -v -count=1 -timeout 5m ./server/
func TestReproLatencyRamp(t *testing.T) {
	opts := DefaultTestOptions
	opts.Port = -1
	opts.JetStream = true
	opts.StoreDir = t.TempDir()
	s := RunServer(&opts)
	defer s.Shutdown()

	t.Logf("NATS server version: %s", VERSION)

	// Publisher connection.
	ncPub := clientConnectToServer(t, s)
	defer ncPub.Close()
	jsPub, err := ncPub.JetStream()
	if err != nil {
		t.Fatal(err)
	}

	// Create stream with MaxAge to exercise the expiry timer path
	// that changed between v2.12.0 and v2.12.1.
	streamName := "REPRO_LATENCY"
	_, err = jsPub.AddStream(&nats.StreamConfig{
		Name:       streamName,
		Subjects:   []string{"repro.orders.>"},
		MaxAge:     7 * 24 * time.Hour,
		Storage:    nats.FileStorage,
		Replicas:   1,
		Duplicates: 10 * time.Second,
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Logf("Stream %q created (MaxAge=7d, FileStorage, Duplicates=10s)", streamName)

	// Consumer connection using new jetstream API for ordered pull consumer.
	ncSub := clientConnectToServer(t, s)
	defer ncSub.Close()
	js, err := jetstream.New(ncSub)
	if err != nil {
		t.Fatal(err)
	}

	// Ordered pull consumer (AckNone, memory storage) matching user's config.
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	cons, err := js.OrderedConsumer(ctx, streamName, jetstream.OrderedConsumerConfig{
		FilterSubjects: []string{"repro.orders.>"},
	})
	if err != nil {
		t.Fatal(err)
	}

	var (
		mu      sync.Mutex
		samples []time.Duration
	)

	addSample := func(d time.Duration) {
		mu.Lock()
		samples = append(samples, d)
		mu.Unlock()
	}
	flushSamples := func() []time.Duration {
		mu.Lock()
		s := samples
		samples = nil
		mu.Unlock()
		return s
	}

	pct := func(sorted []time.Duration, p float64) time.Duration {
		if len(sorted) == 0 {
			return 0
		}
		idx := max(int(math.Ceil(p/100*float64(len(sorted))))-1, 0)
		return sorted[idx]
	}

	var totalPub, totalCon atomic.Int64
	done := make(chan struct{})

	// Publisher: ~50 msgs/sec spread across 10 sub-subjects.
	go func() {
		payload := make([]byte, 256)
		ticker := time.NewTicker(20 * time.Millisecond)
		defer ticker.Stop()
		i := 0
		for {
			select {
			case <-done:
				return
			case <-ticker.C:
				subj := fmt.Sprintf("repro.orders.input.%d", i%10)
				i++
				if _, err := jsPub.Publish(subj, payload); err != nil {
					continue
				}
				totalPub.Add(1)
			}
		}
	}()

	// Consumer: use Consume callback for lower-latency delivery (no polling).
	cc, err := cons.Consume(func(msg jetstream.Msg) {
		now := time.Now()
		meta, err := msg.Metadata()
		if err != nil {
			return
		}
		if latency := now.Sub(meta.Timestamp); latency > 0 {
			addSample(latency)
		}
		totalCon.Add(1)
	})
	if err != nil {
		t.Fatal(err)
	}
	defer cc.Stop()

	// Report every 5s for 2 minutes.
	const (
		testDuration = 2 * time.Minute
		reportEvery  = 5 * time.Second
		numWindows   = int(testDuration / reportEvery)
	)

	type windowStats struct {
		p95 time.Duration
	}
	allStats := make([]windowStats, 0, numWindows)

	t.Log("")
	t.Log("window |   n   |    avg     |    p50     |    p75     |    p95     |    p99     |    max")
	t.Log("-------+-------+------------+------------+------------+------------+------------+------------")

	for w := range numWindows {
		time.Sleep(reportEvery)
		s := flushSamples()
		if len(s) == 0 {
			t.Logf("%5d  |   0   |     -      |     -      |     -      |     -      |     -      |     -", w+1)
			continue
		}

		sort.Slice(s, func(i, j int) bool { return s[i] < s[j] })
		var sum time.Duration
		for _, v := range s {
			sum += v
		}
		avg := sum / time.Duration(len(s))

		ws := windowStats{p95: pct(s, 95)}
		allStats = append(allStats, ws)

		t.Logf("%5d  | %5d | %10s | %10s | %10s | %10s | %10s | %10s",
			w+1, len(s),
			avg.Truncate(time.Microsecond),
			pct(s, 50).Truncate(time.Microsecond),
			pct(s, 75).Truncate(time.Microsecond),
			ws.p95.Truncate(time.Microsecond),
			pct(s, 99).Truncate(time.Microsecond),
			s[len(s)-1].Truncate(time.Microsecond),
		)
	}

	close(done)
	cancel()
	time.Sleep(200 * time.Millisecond)

	t.Logf("\nTotal published: %d, consumed: %d", totalPub.Load(), totalCon.Load())

	// Check for ramp: compare first quarter p95 vs last quarter p95.
	if len(allStats) >= 8 {
		quarter := len(allStats) / 4
		var earlyP95, lateP95 time.Duration
		for _, ws := range allStats[:quarter] {
			earlyP95 += ws.p95
		}
		earlyP95 /= time.Duration(quarter)
		for _, ws := range allStats[len(allStats)-quarter:] {
			lateP95 += ws.p95
		}
		lateP95 /= time.Duration(quarter)

		t.Logf("Early p95 avg: %s", earlyP95.Truncate(time.Microsecond))
		t.Logf("Late  p95 avg: %s", lateP95.Truncate(time.Microsecond))
	}
}