Sustained append-latency degradation on a full DiscardOld stream under concurrent filtered-consumer scans

mkdir repro8552 && cd repro8552   # put main.go here
go mod init repro8552
go get github.com/nats-io/[email protected]
docker run -d --name nats-8552 -p 4222:4222 nats:2.14.6 -js
go run . -url nats://127.0.0.1:4222 -max-bytes 1073741824

// Reproduces nats-server #8552: sustained append-latency degradation on a full
// DiscardOld file stream when filtered consumers read the retention edge concurrently.
//
// It creates a file stream at a byte cap with discard=old and S2 compression, fills it
// past the cap so eviction runs on every append, attaches filtered durable consumers that
// read from the trimmed start (the retention edge, so every fetch force-expires head
// blocks), then measures synchronous-publish (append-commit) latency across many concurrent
// publishers with the consumers RUNNING vs STOPPED. Only the consumers change between the two
// measurements; the publish path is identical.
//
//	go run . [-url nats://127.0.0.1:4222] [-max-bytes 1073741824] [-subjects 256]
//	         [-consumers 8] [-publishers 16] [-measure 30000]
package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"math/rand"
	"sort"
	"sync"
	"sync/atomic"
	"time"

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

const stream = "repro8552"

func main() {
	url := flag.String("url", nats.DefaultURL, "NATS URL")
	maxBytes := flag.Int64("max-bytes", 1<<30, "stream byte cap (also sets ~cap/8MiB blocks)")
	subjects := flag.Int("subjects", 256, "number of subjects log.track.0..N-1")
	consumers := flag.Int("consumers", 8, "filtered edge-reading durable consumers")
	publishers := flag.Int("publishers", 16, "concurrent synchronous publishers during measure")
	measureMsgs := flag.Int("measure", 30000, "sync publishes per measurement phase")
	flag.Parse()

	nc, err := nats.Connect(*url, nats.Timeout(10*time.Second))
	if err != nil {
		log.Fatalf("connect: %v", err)
	}
	defer nc.Drain()
	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatalf("jetstream: %v", err)
	}
	ctx := context.Background()

	subj := make([]string, *subjects)
	for i := range subj {
		subj[i] = fmt.Sprintf("x.%d", i)
	}

	_ = js.DeleteStream(ctx, stream)
	_, err = js.CreateStream(ctx, jetstream.StreamConfig{
		Name:         stream,
		Subjects:     []string{"x.*"},
		Storage:      jetstream.FileStorage,
		Retention:    jetstream.LimitsPolicy,
		Discard:      jetstream.DiscardOld,
		MaxBytes:     *maxBytes,
		Compression:  jetstream.S2Compression,
		Duplicates:   2 * time.Minute,
		MaxConsumers: -1,
	})
	if err != nil {
		log.Fatalf("create stream: %v", err)
	}
	defer js.DeleteStream(context.Background(), stream)

	payload := make([]byte, 1024)
	rand.Read(payload)

	// Fill ~1.4x the cap so the stream is at cap and evicting, with aged blocks.
	fillN := int(*maxBytes/1024) * 14 / 10
	log.Printf("filling %d msgs (~1.4x cap) to age the stream past its %d MiB cap ...", fillN, *maxBytes>>20)
	fill(ctx, js, subj, payload, fillN)
	info, _ := js.Stream(ctx, stream)
	st, _ := info.Info(ctx)
	log.Printf("stream at cap: msgs=%d bytes=%d/%d blocks~=%d",
		st.State.Msgs, st.State.Bytes, *maxBytes, *maxBytes/(8<<20))

	// Filtered durable consumers, deliver-all, reading from the trimmed start.
	cons := make([]jetstream.Consumer, *consumers)
	for i := 0; i < *consumers; i++ {
		filters := []string{}
		for p := i; p < *subjects; p += *consumers {
			filters = append(filters, fmt.Sprintf("x.%d", p))
		}
		c, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{
			Durable:        fmt.Sprintf("consumer%d", i),
			AckPolicy:      jetstream.AckAllPolicy,
			DeliverPolicy:  jetstream.DeliverAllPolicy,
			FilterSubjects: filters,
			MaxAckPending:  4000,
		})
		if err != nil {
			log.Fatalf("consumer %d: %v", i, err)
		}
		cons[i] = c
	}

	fmt.Println("\n=== measurement A: consumers RUNNING ===")
	stop := startConsumers(cons)
	time.Sleep(2 * time.Second)
	a := measure(js, subj, payload, *publishers, *measureMsgs)
	stop()

	fmt.Println("\n=== measurement B: consumers STOPPED ===")
	time.Sleep(4 * time.Second) // let the read side quiesce
	b := measure(js, subj, payload, *publishers, *measureMsgs)

	fmt.Printf("\n%-22s %12s %12s\n", "append latency", "consumers UP", "consumers DOWN")
	fmt.Printf("%-22s %10.2fms %10.2fms\n", "p50", a.p50, b.p50)
	fmt.Printf("%-22s %10.2fms %10.2fms\n", "p99", a.p99, b.p99)
	fmt.Printf("%-22s %10.2fms %10.2fms\n", "max", a.max, b.max)
	fmt.Printf("%-22s %12.0f %12.0f\n", "throughput msgs/s", a.rate, b.rate)
	fmt.Printf("\np50 ratio (UP/DOWN): %.1fx\n", a.p50/b.p50)
}

func fill(ctx context.Context, js jetstream.JetStream, subj []string, payload []byte, n int) {
	var wg sync.WaitGroup
	var next int64
	for w := 0; w < 16; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				i := int(atomic.AddInt64(&next, 1))
				if i > n {
					return
				}
				// async publish; ignore individual acks, we only need them stored
				if _, err := js.PublishAsync(subj[i%len(subj)], payload); err != nil {
					time.Sleep(time.Millisecond)
				}
			}
		}()
	}
	wg.Wait()
	select {
	case <-js.PublishAsyncComplete():
	case <-time.After(60 * time.Second):
	}
}

// startConsumers drives each consumer in a tight fetch loop; returns a stop func.
func startConsumers(cons []jetstream.Consumer) func() {
	done := make(chan struct{})
	var wg sync.WaitGroup
	for _, c := range cons {
		wg.Add(1)
		go func(c jetstream.Consumer) {
			defer wg.Done()
			for {
				select {
				case <-done:
					return
				default:
				}
				batch, err := c.Fetch(500, jetstream.FetchMaxWait(500*time.Millisecond))
				if err != nil {
					continue
				}
				n := 0
				var last jetstream.Msg
				for m := range batch.Messages() {
					last = m
					n++
				}
				if last != nil {
					last.Ack() // AckAll: commit the batch, keep the cursor moving at the edge
				}
			}
		}(c)
	}
	return func() { close(done); wg.Wait() }
}

type stats struct{ p50, p99, max, rate float64 }

func measure(js jetstream.JetStream, subj []string, payload []byte, publishers, total int) stats {
	lat := make([]float64, total)
	var idx int64 = -1
	var wg sync.WaitGroup
	start := time.Now()
	for p := 0; p < publishers; p++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				i := int(atomic.AddInt64(&idx, 1))
				if i >= total {
					return
				}
				t := time.Now()
				_, err := js.Publish(context.Background(), subj[i%len(subj)], payload)
				d := time.Since(t)
				if err != nil {
					lat[i] = -1
					continue
				}
				lat[i] = float64(d.Microseconds()) / 1000.0 // ms
			}
		}()
	}
	wg.Wait()
	elapsed := time.Since(start).Seconds()

	ok := lat[:0]
	for _, v := range lat {
		if v >= 0 {
			ok = append(ok, v)
		}
	}
	sort.Float64s(ok)
	pct := func(p float64) float64 {
		if len(ok) == 0 {
			return 0
		}
		return ok[int(float64(len(ok)-1)*p)]
	}
	return stats{p50: pct(0.50), p99: pct(0.99), max: pct(1.0), rate: float64(len(ok)) / elapsed}
}