chore: add nats benchmarking pkg (#26396)

coderd/x/nats/natsbench/doc.go

@@ -0,0 +1,42 @@
+// Command natsbench benchmarks Coder's NATS-backed pubsub
+// (github.com/coder/coder/v2/coderd/x/nats) under high fan-out load.
+//
+// A run publishes a configurable total number of messages across a set
+// of publishers, subjects, subscribers, and replica nodes, then reports
+// two logical throughput metrics:
+//
+//   - Pubs/sec: messages published divided by the publish duration.
+//   - Deliveries/sec: messages delivered divided by the delivery
+//     duration. Every message is delivered to every subscriber on its
+//     subject, so fan-out makes deliveries exceed publishes. This is
+//     logical throughput, not physical bandwidth.
+//
+// Correctness rules:
+//
+//   - Production defaults: the benchmark measures the pubsub as Coder
+//     configures it. It does not autotune queue or pending limits to
+//     avoid drops; sane defaults are the thing under test.
+//   - Drops as a metric: each subscriber's expected delivery count is
+//     computed up front, so the exact, complete loss is Expected minus
+//     Delivered. Dropped messages are reported in the Drops column, not
+//     treated as a failure; a run that drops still reports the
+//     throughput it achieved. Only a real error (publish failure,
+//     cancellation, hard timeout) invalidates a run.
+//   - Delivery completion: a zero-drop run finishes precisely when every
+//     subscriber reaches its expected count. A run that dropped messages
+//     can never reach that count, so the deliver phase instead completes
+//     by quiescence: once the delivery counter stays flat for a fixed
+//     settle window, the phase ends, and the rate is measured to the
+//     last observed delivery so the idle wait stays out of it.
+//   - Readiness gate: in multi-replica runs, cross-route delivery is
+//     silent when subscription interest has not yet propagated. Before
+//     the measured phase, in-band probe messages prove that every
+//     publisher node can reach every subscriber on its subjects.
+//   - Bounded phases: every wait is bounded by Config.Timeout and fails
+//     with diagnostics (per-subscriber shortfalls, per-node server
+//     stats, goroutine dump) instead of hanging.
+//
+// The command runs the default scenario matrix, one named scenario, or
+// a custom shape, and renders the grouped markdown report to stdout.
+// The heavy benchmarks only run through this command, never in CI.
+package main

coderd/x/nats/natsbench/main.go

@@ -0,0 +1,199 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"golang.org/x/xerrors"
+
+	"cdr.dev/slog/v3"
+	"cdr.dev/slog/v3/sloggers/sloghuman"
+)
+
+func main() {
+	// Cancel running scenarios on SIGINT/SIGTERM so the per-phase
+	// selects unwind cleanly instead of requiring kill -9. stop() is
+	// called explicitly rather than deferred so os.Exit below does not
+	// skip it.
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	err := runCLI(ctx, os.Args, os.Stdout, os.Stderr)
+	stop()
+	if err != nil {
+		_, _ = fmt.Fprintf(os.Stderr, "natsbench: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+// runCLI runs the natsbench command-line interface. args is the full
+// argument vector (args[0] is the program name). The grouped markdown
+// report is written to stdout and progress and logs to stderr. With no
+// flags it runs the default scenario matrix; -scenario runs one named
+// scenario; any custom shape flag runs a single custom configuration.
+// It returns an error when the flags are invalid or any run fails.
+func runCLI(ctx context.Context, args []string, stdout, stderr io.Writer) error {
+	fs := flag.NewFlagSet(args[0], flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var (
+		run     cliRun
+		list    = fs.Bool("list", false, "list default scenarios and exit")
+		verbose = fs.Bool("v", false, "enable debug logging")
+	)
+	fs.StringVar(&run.scenarioName, "scenario", "", "run one named default scenario (see -list)")
+	fs.IntVar(&run.messages, "messages", 0, "total messages across all publishers (0 keeps scenario defaults, or 100000 for custom runs)")
+	fs.IntVar(&run.payload, "payload", Payload8KB, "payload size in bytes (custom run)")
+	fs.IntVar(&run.subjects, "subjects", 10, "number of subjects (custom run)")
+	fs.IntVar(&run.publishers, "publishers", 10, "number of publishers (custom run)")
+	fs.IntVar(&run.subscribers, "subscribers", 50, "number of subscribers (custom run)")
+	fs.IntVar(&run.replicas, "replicas", 1, "number of embedded pubsub nodes (custom run)")
+	fs.IntVar(&run.publishConns, "publish-conns", DefaultConns, "publisher connection pool size (applies to every run)")
+	fs.IntVar(&run.subscribeConns, "subscribe-conns", DefaultConns, "subscriber connection pool size (applies to every run)")
+	fs.Int64Var(&run.seed, "seed", DefaultSeed, "seed for pseudorandom node placement (applies to every run); same seed reproduces the same placement")
+	fs.DurationVar(&run.timeout, "timeout", 2*time.Minute, "per-phase timeout")
+	if err := fs.Parse(args[1:]); err != nil {
+		if errors.Is(err, flag.ErrHelp) {
+			return nil
+		}
+		return err
+	}
+	fs.Visit(func(f *flag.Flag) {
+		switch f.Name {
+		case "payload", "subjects", "publishers", "subscribers", "replicas":
+			run.shapeFlagSet = true
+		}
+	})
+
+	if *list {
+		for _, sc := range DefaultScenarios() {
+			c := sc.Config
+			_, _ = fmt.Fprintf(stdout, "%s: messages=%d payload=%d subjects=%d publishers=%d subscribers=%d replicas=%d\n",
+				sc.Name, c.Messages, c.PayloadSize, c.Subjects, c.Publishers, c.Subscribers, c.Replicas)
+		}
+		return nil
+	}
+
+	scenarios, err := run.scenarios()
+	if err != nil {
+		return err
+	}
+
+	level := slog.LevelWarn
+	if *verbose {
+		level = slog.LevelDebug
+	}
+	logger := slog.Make(sloghuman.Sink(stderr)).Leveled(level)
+
+	// Scenarios run sequentially so they never compete for CPU, memory,
+	// or the network stack and skew each other's numbers.
+	results := make([]ScenarioResult, 0, len(scenarios))
+	failed := false
+	for _, sc := range scenarios {
+		// Stop launching scenarios once interrupted, rather than
+		// letting each remaining Run fail with a confusing topology
+		// error from the canceled context.
+		if err := ctx.Err(); err != nil {
+			return xerrors.Errorf("interrupted before %s: %w", sc.Name, err)
+		}
+		_, _ = fmt.Fprintf(stderr, "running %s...\n", sc.Name)
+		res, runErr := Run(ctx, logger, sc.Config)
+		results = append(results, ScenarioResult{Scenario: sc, Result: res, Err: runErr})
+		if runErr != nil {
+			failed = true
+			_, _ = fmt.Fprintf(stderr, "%s failed: %v\n", sc.Name, runErr)
+			continue
+		}
+		_, _ = fmt.Fprintf(stderr, "%s: pubs/sec=%.0f deliveries/sec=%.0f\n",
+			sc.Name, res.PubsPerSec, res.DeliveriesPerSec)
+	}
+
+	if err := RenderMarkdown(stdout, results); err != nil {
+		return xerrors.Errorf("render report: %w", err)
+	}
+	if failed {
+		return xerrors.New("one or more runs failed; see report")
+	}
+	return nil
+}
+
+// cliRun holds the parsed flag values that select the scenarios to run.
+type cliRun struct {
+	scenarioName string
+	// shapeFlagSet is true when any custom-shape flag was passed, which
+	// selects a single custom run.
+	shapeFlagSet   bool
+	messages       int
+	payload        int
+	subjects       int
+	publishers     int
+	subscribers    int
+	replicas       int
+	publishConns   int
+	subscribeConns int
+	seed           int64
+	timeout        time.Duration
+}
+
+// scenarios resolves the parsed flags into the scenarios to run: one
+// named default scenario, a single custom shape when any shape flag was
+// set, or the full default matrix. The per-phase timeout is applied to
+// every scenario, and -messages overrides the message count when set.
+func (c cliRun) scenarios() ([]Scenario, error) {
+	switch {
+	case c.scenarioName != "":
+		if c.shapeFlagSet {
+			return nil, xerrors.New("-scenario and custom shape flags are mutually exclusive")
+		}
+		for _, sc := range DefaultScenarios() {
+			if sc.Name != c.scenarioName {
+				continue
+			}
+			if c.messages > 0 {
+				sc.Config.Messages = c.messages
+			}
+			sc.Config.PublishConns = c.publishConns
+			sc.Config.SubscribeConns = c.subscribeConns
+			sc.Config.Seed = c.seed
+			sc.Config.Timeout = c.timeout
+			return []Scenario{sc}, nil
+		}
+		return nil, xerrors.Errorf("unknown scenario %q; use -list to see available scenarios", c.scenarioName)
+	case c.shapeFlagSet:
+		messages := c.messages
+		if messages <= 0 {
+			messages = DefaultMessages
+		}
+		return []Scenario{{
+			Name: "custom",
+			Config: Config{
+				Messages:       messages,
+				PayloadSize:    c.payload,
+				Subjects:       c.subjects,
+				Publishers:     c.publishers,
+				Subscribers:    c.subscribers,
+				Replicas:       c.replicas,
+				PublishConns:   c.publishConns,
+				SubscribeConns: c.subscribeConns,
+				Seed:           c.seed,
+				Timeout:        c.timeout,
+			},
+		}}, nil
+	default:
+		scenarios := DefaultScenarios()
+		for i := range scenarios {
+			if c.messages > 0 {
+				scenarios[i].Config.Messages = c.messages
+			}
+			scenarios[i].Config.PublishConns = c.publishConns
+			scenarios[i].Config.SubscribeConns = c.subscribeConns
+			scenarios[i].Config.Seed = c.seed
+			scenarios[i].Config.Timeout = c.timeout
+		}
+		return scenarios, nil
+	}
+}

coderd/x/nats/natsbench/main_internal_test.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/coder/coder/v2/testutil"
+)
+
+func TestCLIScenarios(t *testing.T) {
+	t.Parallel()
+
+	t.Run("DefaultMatrix", func(t *testing.T) {
+		t.Parallel()
+		got, err := cliRun{timeout: testutil.WaitShort, publishConns: DefaultConns, subscribeConns: DefaultConns}.scenarios()
+		require.NoError(t, err)
+		require.Len(t, got, len(DefaultScenarios()))
+		for _, sc := range got {
+			require.Equal(t, testutil.WaitShort, sc.Config.Timeout)
+			require.Equal(t, DefaultConns, sc.Config.PublishConns)
+			require.Equal(t, DefaultConns, sc.Config.SubscribeConns)
+		}
+	})
+
+	t.Run("ConnOverride", func(t *testing.T) {
+		t.Parallel()
+		got, err := cliRun{timeout: testutil.WaitShort, publishConns: 1, subscribeConns: 1}.scenarios()
+		require.NoError(t, err)
+		for _, sc := range got {
+			require.Equal(t, 1, sc.Config.PublishConns)
+			require.Equal(t, 1, sc.Config.SubscribeConns)
+		}
+	})
+
+	t.Run("MessageOverride", func(t *testing.T) {
+		t.Parallel()
+		got, err := cliRun{messages: 5, timeout: testutil.WaitShort}.scenarios()
+		require.NoError(t, err)
+		for _, sc := range got {
+			require.Equal(t, 5, sc.Config.Messages)
+		}
+	})
+
+	t.Run("NamedScenario", func(t *testing.T) {
+		t.Parallel()
+		got, err := cliRun{scenarioName: "8KiB-1r", messages: 9, timeout: testutil.WaitShort}.scenarios()
+		require.NoError(t, err)
+		require.Len(t, got, 1)
+		require.Equal(t, "8KiB-1r", got[0].Name)
+		require.Equal(t, 9, got[0].Config.Messages)
+	})
+
+	t.Run("UnknownScenario", func(t *testing.T) {
+		t.Parallel()
+		_, err := cliRun{scenarioName: "nope", timeout: testutil.WaitShort}.scenarios()
+		require.Error(t, err)
+	})
+
+	t.Run("CustomShape", func(t *testing.T) {
+		t.Parallel()
+		got, err := cliRun{
+			shapeFlagSet:   true,
+			payload:        Payload64KB,
+			subjects:       3,
+			publishers:     4,
+			subscribers:    8,
+			replicas:       2,
+			publishConns:   DefaultConns,
+			subscribeConns: DefaultConns,
+			timeout:        testutil.WaitShort,
+		}.scenarios()
+		require.NoError(t, err)
+		require.Len(t, got, 1)
+		require.Equal(t, "custom", got[0].Name)
+		// Custom runs default to the standard message count.
+		require.Equal(t, DefaultMessages, got[0].Config.Messages)
+		require.Equal(t, Payload64KB, got[0].Config.PayloadSize)
+		require.Equal(t, 2, got[0].Config.Replicas)
+		require.Equal(t, DefaultConns, got[0].Config.PublishConns)
+	})
+
+	t.Run("ScenarioAndShapeConflict", func(t *testing.T) {
+		t.Parallel()
+		_, err := cliRun{scenarioName: "8KiB-1r", shapeFlagSet: true, timeout: testutil.WaitShort}.scenarios()
+		require.Error(t, err)
+	})
+}

coderd/x/nats/natsbench/natsbench.go

@@ -0,0 +1,171 @@
+package main
+
+import (
+	"context"
+	"time"
+
+	natsserver "github.com/nats-io/nats-server/v2/server"
+	"golang.org/x/xerrors"
+
+	"cdr.dev/slog/v3"
+)
+
+const (
+	// DefaultMessages is the default total message count for a run.
+	DefaultMessages = 100000
+	// Payload8KB and Payload64KB are the standard matrix payload sizes.
+	Payload8KB  = 8 << 10
+	Payload64KB = 64 << 10
+)
+
+// Config describes one benchmark run. Run validates a fully populated
+// config and applies no defaults of its own (the CLI fills defaults
+// before calling Run), so every required field must be set. The
+// optional tuning fields below are passed straight through to
+// nats.Options, where the nats package applies its own zero-value
+// defaults. The benchmark deliberately does NOT autotune these knobs:
+// it measures the pubsub as Coder configures it in production, so any
+// dropped messages are reported as a metric rather than tuned away.
+type Config struct {
+	// Messages is the TOTAL number of messages across all publishers,
+	// split evenly with the remainder assigned to publisher 0. Must be
+	// at least 1.
+	Messages int
+	// PayloadSize is the benchmark message size in bytes.
+	PayloadSize int
+	// Subjects is the number of distinct subjects ("bench.<i>").
+	Subjects int
+	// Publishers is the number of concurrent publishers. Publisher i
+	// publishes to subject i % Subjects from node i % Replicas.
+	Publishers int
+	// Subscribers is the number of subscribers. Subscriber j listens on
+	// subject j % Subjects from node j % Replicas.
+	Subscribers int
+	// Replicas is the number of embedded pubsub nodes. 1 runs a single
+	// node; greater values run a fully meshed cluster.
+	Replicas int
+	// Seed selects the pseudorandom node placement. The same seed
+	// reproduces the same placement for a given shape; change it to
+	// sample a different placement. Zero is a valid, deterministic
+	// seed.
+	Seed int64
+
+	// InProcess uses in-process server connections instead of TCP
+	// loopback.
+	InProcess bool
+	// PublishConns and SubscribeConns configure the pubsub connection
+	// pools. Zero passes through to nats.Options, which defaults each
+	// pool to a single connection (the production default).
+	PublishConns   int
+	SubscribeConns int
+	// LocalQueueMsgs overrides the per-subscription NATS pending
+	// message limit. Zero uses the nats package production default;
+	// set it only for sensitivity analysis.
+	LocalQueueMsgs int
+	// LocalQueueBytes overrides the per-subscription NATS pending byte
+	// limit. Zero uses the nats package production default.
+	LocalQueueBytes int
+	// MaxPending overrides the embedded server's per-client outbound
+	// pending byte budget. Zero uses the nats package production
+	// default.
+	MaxPending int64
+	// Timeout bounds each phase (readiness, publish, deliver). It must
+	// be positive.
+	Timeout time.Duration
+}
+
+// Result reports one run's exact accounting and throughput.
+type Result struct {
+	// Config is the fully resolved configuration the run used,
+	// including derived sizing.
+	Config Config
+
+	// Published is the number of successfully published messages.
+	Published int64
+	// Delivered is the number of benchmark messages observed across all
+	// subscribers. Fan-out makes this exceed Published whenever a
+	// subject has multiple subscribers; it is logical throughput, not
+	// physical bandwidth.
+	Delivered int64
+	// Expected is the total number of benchmark deliveries the plan
+	// requires: the sum of every subscriber's expected count. It is the
+	// exact denominator for the drop rate.
+	Expected int64
+	// Drops is Expected minus Delivered: the number of deliveries that
+	// never arrived. It is the authoritative, complete loss count, since
+	// the ErrDroppedMessages signal coalesces and cross-node routed loss
+	// is silent. Drops are a reported metric, not a failure: a run with
+	// drops still produces trustworthy throughput numbers for what it did
+	// deliver.
+	Drops int64
+
+	// ConvergenceDuration is how long the readiness gate took to
+	// propagate subscription interest across the cluster, measured from
+	// the first probe (right after all subscriptions are registered) to
+	// full propagation. Zero for single-node runs, which need no gate.
+	ConvergenceDuration time.Duration
+
+	// PublishDuration spans the hot start to the last publisher
+	// finishing, including the final flush. DeliverDuration spans the
+	// hot start to the last delivery: for a zero-drop run that is the
+	// last subscriber reaching its expected count; for a run that dropped
+	// messages it is the last observed counter progress before the
+	// settle window elapsed, so the settle delay stays out of the rate.
+	PublishDuration time.Duration
+	DeliverDuration time.Duration
+
+	// PubsPerSec is Published / PublishDuration. DeliveriesPerSec is
+	// Delivered / DeliverDuration.
+	PubsPerSec       float64
+	DeliveriesPerSec float64
+}
+
+// validate rejects configurations the engine cannot run.
+func (c Config) validate() error {
+	if c.Messages < 1 {
+		return xerrors.Errorf("messages must be at least 1, got %d", c.Messages)
+	}
+	if c.PayloadSize < 1 {
+		return xerrors.Errorf("payload size must be at least 1, got %d", c.PayloadSize)
+	}
+	if c.PayloadSize > natsserver.MAX_PAYLOAD_SIZE {
+		return xerrors.Errorf("payload size %d exceeds the NATS max payload %d", c.PayloadSize, natsserver.MAX_PAYLOAD_SIZE)
+	}
+	if c.Subjects < 1 {
+		return xerrors.Errorf("subjects must be at least 1, got %d", c.Subjects)
+	}
+	if c.Publishers < 1 {
+		return xerrors.Errorf("publishers must be at least 1, got %d", c.Publishers)
+	}
+	if c.Subscribers < 1 {
+		return xerrors.Errorf("subscribers must be at least 1, got %d", c.Subscribers)
+	}
+	if c.Replicas < 1 {
+		return xerrors.Errorf("replicas must be at least 1, got %d", c.Replicas)
+	}
+	if c.Timeout <= 0 {
+		return xerrors.Errorf("timeout must be positive, got %s", c.Timeout)
+	}
+	return nil
+}
+
+// Run executes one benchmark run: build the deterministic plan, start
+// the topology, and drive the workload. The config must be fully
+// populated; Run applies no defaults. On failure it returns any partial
+// Result alongside the error for diagnostics. Dropped messages are not
+// a failure: a successful Result may still report a nonzero Drops count.
+func Run(ctx context.Context, logger slog.Logger, cfg Config) (*Result, error) {
+	if err := cfg.validate(); err != nil {
+		return nil, xerrors.Errorf("validate config: %w", err)
+	}
+
+	pl := buildPlan(cfg)
+
+	top, err := buildTopology(ctx, logger, cfg)
+	if err != nil {
+		return nil, xerrors.Errorf("build topology: %w", err)
+	}
+	defer top.closeAll()
+
+	return runWorkload(ctx, logger, top, pl, cfg)
+}

coderd/x/nats/natsbench/natsbench_internal_test.go

@@ -0,0 +1,106 @@
+package main
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/coder/coder/v2/testutil"
+)
+
+func TestConfigValidate(t *testing.T) {
+	t.Parallel()
+
+	base := Config{
+		Messages: 100, PayloadSize: 1024, Subjects: 1, Publishers: 1, Subscribers: 1, Replicas: 1,
+		Timeout: testutil.WaitShort,
+	}
+	require.NoError(t, base.validate())
+
+	cases := []struct {
+		name   string
+		mutate func(*Config)
+	}{
+		{"NoPayload", func(c *Config) { c.PayloadSize = 0 }},
+		{"OversizedPayload", func(c *Config) { c.PayloadSize = 2 << 20 }},
+		{"NoSubjects", func(c *Config) { c.Subjects = 0 }},
+		{"NoPublishers", func(c *Config) { c.Publishers = 0 }},
+		{"NoSubscribers", func(c *Config) { c.Subscribers = 0 }},
+		{"NoReplicas", func(c *Config) { c.Replicas = 0 }},
+		{"NegativeMessages", func(c *Config) { c.Messages = -1 }},
+		{"NoTimeout", func(c *Config) { c.Timeout = 0 }},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			cfg := base
+			tc.mutate(&cfg)
+			require.Error(t, cfg.validate())
+		})
+	}
+}
+
+func TestRunSingleNode(t *testing.T) {
+	t.Parallel()
+
+	ctx := testutil.Context(t, testutil.WaitLong)
+	logger := testutil.Logger(t)
+
+	cfg := Config{
+		Messages:    1000,
+		PayloadSize: 1024,
+		Subjects:    2,
+		Publishers:  4,
+		Subscribers: 8,
+		Replicas:    1,
+		InProcess:   true,
+		Timeout:     testutil.WaitLong,
+	}
+	res, err := Run(ctx, logger, cfg)
+	require.NoError(t, err)
+
+	pl := buildPlan(cfg)
+	require.EqualValues(t, cfg.Messages, res.Published)
+	require.EqualValues(t, pl.totalExpected, res.Expected)
+	require.EqualValues(t, pl.totalExpected, res.Delivered)
+	require.Greater(t, res.Delivered, res.Published, "fan-out must exceed publishes")
+	require.Zero(t, res.Drops)
+	require.Greater(t, res.PubsPerSec, 0.0)
+	require.Greater(t, res.DeliveriesPerSec, 0.0)
+	require.Greater(t, res.DeliverDuration, time.Duration(0))
+	require.GreaterOrEqual(t, res.DeliverDuration, res.PublishDuration)
+}
+
+func TestRunCluster(t *testing.T) {
+	t.Parallel()
+	if testing.Short() {
+		t.SkipNow()
+	}
+
+	ctx := testutil.Context(t, testutil.WaitSuperLong)
+	logger := testutil.Logger(t)
+
+	// Random node placement across 3 replicas makes cross-node
+	// delivery the common case, exercising route propagation through
+	// the readiness gate.
+	cfg := Config{
+		Messages:    600,
+		PayloadSize: 512,
+		Subjects:    2,
+		Publishers:  4,
+		Subscribers: 6,
+		Replicas:    3,
+		Timeout:     testutil.WaitLong,
+	}
+	res, err := Run(ctx, logger, cfg)
+	require.NoError(t, err)
+
+	pl := buildPlan(cfg)
+	require.EqualValues(t, cfg.Messages, res.Published)
+	require.EqualValues(t, pl.totalExpected, res.Expected)
+	require.EqualValues(t, pl.totalExpected, res.Delivered)
+	require.Zero(t, res.Drops)
+	require.Greater(t, res.PubsPerSec, 0.0)
+	require.Greater(t, res.DeliveriesPerSec, 0.0)
+}

coderd/x/nats/natsbench/plan.go

@@ -0,0 +1,122 @@
+package main
+
+import (
+	"fmt"
+	"math/rand/v2"
+	"slices"
+)
+
+// plan assigns each publisher and subscriber to a subject round-robin by
+// index (index % Subjects), places each one on a pseudorandom replica
+// node (seeded for reproducibility), and precomputes each subscriber's
+// expected delivery count so the workload can do exact accounting.
+//
+// Node placement is random to model a real deployment, where each
+// client connects to an arbitrary replica. Round-robin node placement
+// would instead co-locate publisher i and subscriber i on the same
+// node and, whenever Replicas divides Subjects, keep a subject's entire
+// traffic local, which understates cross-node routing cost.
+//
+// Worked example with Subjects=2, Publishers=3, Subscribers=4,
+// Messages=100. Subject assignment is round-robin (index % Subjects)
+// and the 100 messages split 34/33/33 (the remainder lands on publisher
+// 0):
+//
+//	publisher 0 -> subject 0, sends 34
+//	publisher 1 -> subject 1, sends 33
+//	publisher 2 -> subject 0, sends 33
+//
+//	subject 0 receives 34+33 = 67 (publishers 0 and 2)
+//	subject 1 receives 33      (publisher 1)
+//
+//	subscriber 0 -> subject 0, expects 67
+//	subscriber 1 -> subject 1, expects 33
+//	subscriber 2 -> subject 0, expects 67
+//	subscriber 3 -> subject 1, expects 33
+//
+// totalExpected = 67+33+67+33 = 200. It exceeds the 100 published
+// messages because each subject has two subscribers, so every message
+// is delivered twice. That fan-out is the logical delivery count the
+// benchmark measures. Each publisher's and subscriber's node is drawn
+// independently from [0, Replicas).
+type plan struct {
+	// perPubMsgs[i] is the number of messages publisher i sends.
+	perPubMsgs []int
+	// pubSubject[i] is publisher i's subject index.
+	pubSubject []int
+	// pubNode[i] is publisher i's node index.
+	pubNode []int
+	// subSubject[j] is subscriber j's subject index.
+	subSubject []int
+	// subNode[j] is subscriber j's node index.
+	subNode []int
+	// expectPerSub[j] is how many benchmark messages subscriber j must
+	// observe: the total sent to its subject.
+	expectPerSub []int
+	// totalExpected is the sum of expectPerSub.
+	totalExpected int
+	// pubNodes and subNodes are the sorted distinct node indexes that
+	// host at least one publisher or subscriber. Several publishers or
+	// subscribers can share a node, so these dedupe pubNode / subNode
+	// for callers that act once per node (for example flushing).
+	pubNodes []int
+	subNodes []int
+}
+
+// buildPlan computes the workload assignment for cfg. Subject and
+// message assignment is deterministic; node placement is pseudorandom
+// but fully determined by cfg.Seed, so a given config reproduces the
+// same plan. cfg must already be validated: all counts are at least 1.
+func buildPlan(cfg Config) plan {
+	pl := plan{
+		perPubMsgs:   make([]int, cfg.Publishers),
+		pubSubject:   make([]int, cfg.Publishers),
+		pubNode:      make([]int, cfg.Publishers),
+		subSubject:   make([]int, cfg.Subscribers),
+		subNode:      make([]int, cfg.Subscribers),
+		expectPerSub: make([]int, cfg.Subscribers),
+	}
+
+	// Seed both PCG streams from cfg.Seed so placement is reproducible
+	// for a given seed. Node placement is benchmark randomness, not
+	// security-sensitive.
+	seed := uint64(cfg.Seed)                 //nolint:gosec // G115: deterministic benchmark seed, sign irrelevant.
+	rng := rand.New(rand.NewPCG(seed, seed)) //nolint:gosec // G404: placement randomness, not security-sensitive.
+
+	base := cfg.Messages / cfg.Publishers
+	remainder := cfg.Messages % cfg.Publishers
+	perSubjectMsgs := make([]int, cfg.Subjects)
+	for i := range cfg.Publishers {
+		msgs := base
+		if i == 0 {
+			msgs += remainder
+		}
+		pl.perPubMsgs[i] = msgs
+		pl.pubSubject[i] = i % cfg.Subjects
+		pl.pubNode[i] = rng.IntN(cfg.Replicas)
+		perSubjectMsgs[pl.pubSubject[i]] += msgs
+	}
+
+	for j := range cfg.Subscribers {
+		pl.subSubject[j] = j % cfg.Subjects
+		pl.subNode[j] = rng.IntN(cfg.Replicas)
+		pl.expectPerSub[j] = perSubjectMsgs[pl.subSubject[j]]
+		pl.totalExpected += pl.expectPerSub[j]
+	}
+
+	pl.pubNodes = uniqueInts(pl.pubNode)
+	pl.subNodes = uniqueInts(pl.subNode)
+	return pl
+}
+
+// uniqueInts returns the sorted distinct values of a slice.
+func uniqueInts(values []int) []int {
+	out := slices.Clone(values)
+	slices.Sort(out)
+	return slices.Compact(out)
+}
+
+// subjectName returns the NATS subject for subject index i.
+func subjectName(i int) string {
+	return fmt.Sprintf("bench.%d", i)
+}

coderd/x/nats/natsbench/plan_internal_test.go

@@ -0,0 +1,165 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestBuildPlan(t *testing.T) {
+	t.Parallel()
+
+	// Node placement is random, so these cases assert only the
+	// deterministic subject and message assignment.
+	cases := []struct {
+		name              string
+		cfg               Config
+		wantPerPubMsgs    []int
+		wantPubSubject    []int
+		wantSubSubject    []int
+		wantExpectPerSub  []int
+		wantTotalExpected int
+	}{
+		{
+			name: "EvenSplit",
+			cfg: Config{
+				Messages: 100, Publishers: 4, Subjects: 2, Subscribers: 4, Replicas: 2,
+			},
+			wantPerPubMsgs:    []int{25, 25, 25, 25},
+			wantPubSubject:    []int{0, 1, 0, 1},
+			wantSubSubject:    []int{0, 1, 0, 1},
+			wantExpectPerSub:  []int{50, 50, 50, 50},
+			wantTotalExpected: 200,
+		},
+		{
+			name: "RemainderToPublisherZero",
+			cfg: Config{
+				Messages: 10, Publishers: 3, Subjects: 3, Subscribers: 3, Replicas: 1,
+			},
+			wantPerPubMsgs:    []int{4, 3, 3},
+			wantPubSubject:    []int{0, 1, 2},
+			wantSubSubject:    []int{0, 1, 2},
+			wantExpectPerSub:  []int{4, 3, 3},
+			wantTotalExpected: 10,
+		},
+		{
+			name: "MorePublishersThanSubjects",
+			cfg: Config{
+				Messages: 30, Publishers: 5, Subjects: 2, Subscribers: 2, Replicas: 3,
+			},
+			// Publisher 0 gets 6 + remainder 0; 30/5 = 6 each.
+			wantPerPubMsgs: []int{6, 6, 6, 6, 6},
+			wantPubSubject: []int{0, 1, 0, 1, 0},
+			wantSubSubject: []int{0, 1},
+			// Subject 0 receives from publishers 0, 2, 4 (18 msgs);
+			// subject 1 from publishers 1, 3 (12 msgs).
+			wantExpectPerSub:  []int{18, 12},
+			wantTotalExpected: 30,
+		},
+		{
+			name: "SubscriberOnSubjectWithoutPublishers",
+			cfg: Config{
+				Messages: 9, Publishers: 1, Subjects: 2, Subscribers: 4, Replicas: 1,
+			},
+			wantPerPubMsgs: []int{9},
+			wantPubSubject: []int{0},
+			wantSubSubject: []int{0, 1, 0, 1},
+			// Subscribers on subject 1 expect nothing.
+			wantExpectPerSub:  []int{9, 0, 9, 0},
+			wantTotalExpected: 18,
+		},
+		{
+			name: "FanOutExceedsPublishes",
+			cfg: Config{
+				Messages: 100, Publishers: 2, Subjects: 1, Subscribers: 10, Replicas: 5,
+			},
+			wantPerPubMsgs:    []int{50, 50},
+			wantPubSubject:    []int{0, 0},
+			wantSubSubject:    []int{0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
+			wantExpectPerSub:  []int{100, 100, 100, 100, 100, 100, 100, 100, 100, 100},
+			wantTotalExpected: 1000,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			pl := buildPlan(tc.cfg)
+			require.Equal(t, tc.wantPerPubMsgs, pl.perPubMsgs)
+			require.Equal(t, tc.wantPubSubject, pl.pubSubject)
+			require.Equal(t, tc.wantSubSubject, pl.subSubject)
+			require.Equal(t, tc.wantExpectPerSub, pl.expectPerSub)
+			require.Equal(t, tc.wantTotalExpected, pl.totalExpected)
+
+			// Every node index is a valid replica, and pubNodes/subNodes
+			// are the sorted distinct sets of those indexes.
+			for _, n := range append(append([]int{}, pl.pubNode...), pl.subNode...) {
+				require.GreaterOrEqual(t, n, 0)
+				require.Less(t, n, tc.cfg.Replicas)
+			}
+			require.Equal(t, uniqueInts(pl.pubNode), pl.pubNodes)
+			require.Equal(t, uniqueInts(pl.subNode), pl.subNodes)
+		})
+	}
+}
+
+func TestBuildPlanNodePlacement(t *testing.T) {
+	t.Parallel()
+
+	cfg := Config{Messages: 1000, Publishers: 50, Subjects: 10, Subscribers: 200, Replicas: 7}
+
+	// The same seed reproduces the same placement.
+	require.Equal(t, buildPlan(cfg).pubNode, buildPlan(cfg).pubNode)
+	require.Equal(t, buildPlan(cfg).subNode, buildPlan(cfg).subNode)
+
+	// A different seed (very likely) produces a different placement.
+	other := cfg
+	other.Seed = cfg.Seed + 1
+	require.NotEqual(t, buildPlan(cfg).subNode, buildPlan(other).subNode)
+
+	// With many clients and few replicas, placement spreads across
+	// every node rather than collapsing onto one.
+	pl := buildPlan(cfg)
+	require.Len(t, pl.pubNodes, cfg.Replicas)
+	require.Len(t, pl.subNodes, cfg.Replicas)
+}
+
+// TestDefaultSeedSpreadsEvenly guards the DefaultSeed comment's claim:
+// with the matrix shape (10 publishers) it places publishers perfectly
+// evenly across nodes at every matrix replica count (1, 5, and 10), so
+// no node is left publisher-less and cross-node routing is not skewed.
+func TestDefaultSeedSpreadsEvenly(t *testing.T) {
+	t.Parallel()
+
+	const publishers = 10
+	for _, replicas := range []int{1, 5, 10} {
+		cfg := Config{
+			Messages:   1000,
+			Subjects:   10,
+			Publishers: publishers,
+			// Subscribers do not affect publisher placement, but a plan
+			// needs at least one.
+			Subscribers: 1,
+			Replicas:    replicas,
+			Seed:        DefaultSeed,
+		}
+		pl := buildPlan(cfg)
+
+		perNode := make([]int, replicas)
+		for _, n := range pl.pubNode {
+			perNode[n]++
+		}
+		want := publishers / replicas
+		for node, got := range perNode {
+			require.Equalf(t, want, got,
+				"replicas=%d node=%d expected %d publishers, got %d (placement %v)",
+				replicas, node, want, got, perNode)
+		}
+	}
+}
+
+func TestSubjectName(t *testing.T) {
+	t.Parallel()
+	require.Equal(t, "bench.0", subjectName(0))
+	require.Equal(t, "bench.17", subjectName(17))
+}

coderd/x/nats/natsbench/readiness.go

@@ -0,0 +1,199 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"golang.org/x/xerrors"
+)
+
+const (
+	// probePrefix tags readiness probes so they are distinguishable
+	// from benchmark payloads. Benchmark payloads are all zeros, so any
+	// payload starting with this non-zero ASCII prefix can only be a
+	// probe.
+	probePrefix = "natsbench-probe:"
+	// probeInterval is how often probes are re-published while waiting
+	// for cross-route subscription interest to propagate.
+	probeInterval = 25 * time.Millisecond
+)
+
+// probePrefixBytes avoids per-call allocation in probeNode, which runs
+// on every delivered message.
+var probePrefixBytes = []byte(probePrefix)
+
+// probePayload encodes a readiness probe identifying the publishing
+// node: the probe prefix followed by the node index in decimal ASCII.
+func probePayload(node int) []byte {
+	return []byte(probePrefix + strconv.Itoa(node))
+}
+
+// probeNode decodes a readiness probe. It reports false for benchmark
+// payloads, which are all zeros and never match the probe prefix.
+func probeNode(payload []byte) (int, bool) {
+	rest, ok := bytes.CutPrefix(payload, probePrefixBytes)
+	if !ok {
+		return 0, false
+	}
+	node, err := strconv.Atoi(string(rest))
+	if err != nil {
+		return 0, false
+	}
+	return node, true
+}
+
+// probeTracker records which publisher nodes a subscriber has observed
+// probes from.
+type probeTracker struct {
+	mu   sync.Mutex
+	seen map[int]struct{}
+}
+
+func newProbeTracker() *probeTracker {
+	return &probeTracker{seen: make(map[int]struct{})}
+}
+
+func (t *probeTracker) observe(node int) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.seen[node] = struct{}{}
+}
+
+// missing returns the required node indexes not yet observed, sorted.
+func (t *probeTracker) missing(required map[int]struct{}) []int {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	var out []int
+	for node := range required {
+		if _, ok := t.seen[node]; !ok {
+			out = append(out, node)
+		}
+	}
+	slices.Sort(out)
+	return out
+}
+
+// subjectNodes maps each subject index to the set of publisher node
+// indexes that publish to it. This single mapping drives the whole
+// gate: it is both the probe schedule (each node probes its subjects)
+// and, looked up by a subscriber's subject, that subscriber's required
+// probe set. Subjects without publishers are absent (nothing required).
+func subjectNodes(pl plan) map[int]map[int]struct{} {
+	out := make(map[int]map[int]struct{})
+	for i, node := range pl.pubNode {
+		subject := pl.pubSubject[i]
+		if out[subject] == nil {
+			out[subject] = make(map[int]struct{})
+		}
+		out[subject][node] = struct{}{}
+	}
+	return out
+}
+
+// readinessCheckInterval is how often the gate polls for convergence.
+// It is finer than probeInterval so the measured convergence time
+// reflects when probes actually propagated rather than the probe
+// republish cadence.
+const readinessCheckInterval = time.Millisecond
+
+// awaitTopologyReady proves that subscription interest has propagated to
+// every publisher node before the measured phase, and returns how long
+// that took. Each publisher node repeatedly publishes an in-band probe
+// on every subject it will publish to; the gate converges when every
+// subscriber has observed a probe from every publisher node targeting
+// its subject. Without this, routed deliveries silently undercount on
+// fresh clusters.
+//
+// The returned duration is the cluster convergence time: from the first
+// probe to the moment interest has propagated everywhere. It is
+// measured from gate entry, which is immediately after all subscriptions
+// are registered.
+func awaitTopologyReady(ctx context.Context, top *topology, pl plan, timeout time.Duration, trackers []*probeTracker) (time.Duration, error) {
+	bySubject := subjectNodes(pl)
+	required := make([]map[int]struct{}, len(pl.subSubject))
+	for j, subject := range pl.subSubject {
+		required[j] = bySubject[subject]
+	}
+
+	publishProbes := func() error {
+		for subject, nodes := range bySubject {
+			for node := range nodes {
+				if err := top.nodes[node].Publish(subjectName(subject), probePayload(node)); err != nil {
+					return xerrors.Errorf("publish probe from node %d on %s: %w", node, subjectName(subject), err)
+				}
+			}
+		}
+		for _, node := range pl.pubNodes {
+			if err := top.nodes[node].Flush(); err != nil {
+				return xerrors.Errorf("flush probes from node %d: %w", node, err)
+			}
+		}
+		return nil
+	}
+
+	started := time.Now()
+	deadline := time.NewTimer(timeout)
+	defer deadline.Stop()
+	publishTicker := time.NewTicker(probeInterval)
+	defer publishTicker.Stop()
+	checkTicker := time.NewTicker(readinessCheckInterval)
+	defer checkTicker.Stop()
+
+	if err := publishProbes(); err != nil {
+		return 0, err
+	}
+	if isReady(trackers, required) {
+		return time.Since(started), nil
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return 0, xerrors.Errorf("readiness gate canceled: %w", ctx.Err())
+		case <-deadline.C:
+			return 0, xerrors.Errorf("readiness gate timed out after %s: %s", timeout, unreadySubscribers(trackers, required))
+		case <-publishTicker.C:
+			if err := publishProbes(); err != nil {
+				return 0, err
+			}
+		case <-checkTicker.C:
+			if isReady(trackers, required) {
+				return time.Since(started), nil
+			}
+		}
+	}
+}
+
+func isReady(trackers []*probeTracker, required []map[int]struct{}) bool {
+	for j, tracker := range trackers {
+		if len(tracker.missing(required[j])) > 0 {
+			return false
+		}
+	}
+	return true
+}
+
+// unreadySubscribers describes which subscribers are still missing
+// probes from which publisher nodes.
+func unreadySubscribers(trackers []*probeTracker, required []map[int]struct{}) string {
+	const maxEntries = 20
+	var entries []string
+	for j, tracker := range trackers {
+		missing := tracker.missing(required[j])
+		if len(missing) == 0 {
+			continue
+		}
+		if len(entries) >= maxEntries {
+			entries = append(entries, "...")
+			break
+		}
+		entries = append(entries, fmt.Sprintf("subscriber %d missing probes from nodes %v", j, missing))
+	}
+	return strings.Join(entries, "; ")
+}

coderd/x/nats/natsbench/readiness_internal_test.go

@@ -0,0 +1,82 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestProbeRoundTrip(t *testing.T) {
+	t.Parallel()
+
+	for _, node := range []int{0, 1, 7, 1 << 30} {
+		got, ok := probeNode(probePayload(node))
+		require.True(t, ok)
+		require.Equal(t, node, got)
+	}
+}
+
+func TestProbeNodeRejectsBenchmarkPayloads(t *testing.T) {
+	t.Parallel()
+
+	// Benchmark payloads are all zeros, at any size.
+	for _, size := range []int{1, 2, 9, Payload8KB} {
+		_, ok := probeNode(make([]byte, size))
+		require.False(t, ok)
+	}
+	// The bare prefix has no node index.
+	_, ok := probeNode([]byte(probePrefix))
+	require.False(t, ok)
+	// A trailing non-digit byte fails decoding.
+	_, ok = probeNode(append(probePayload(3), 0))
+	require.False(t, ok)
+}
+
+func TestProbeTracker(t *testing.T) {
+	t.Parallel()
+
+	tracker := newProbeTracker()
+	required := map[int]struct{}{0: {}, 2: {}}
+
+	require.Equal(t, []int{0, 2}, tracker.missing(required))
+	tracker.observe(0)
+	tracker.observe(1) // Not required; ignored by missing.
+	require.Equal(t, []int{2}, tracker.missing(required))
+	tracker.observe(2)
+	require.Empty(t, tracker.missing(required))
+	require.Empty(t, tracker.missing(nil))
+}
+
+func TestSubjectNodes(t *testing.T) {
+	t.Parallel()
+
+	pl := buildPlan(Config{
+		Messages: 30, Publishers: 5, Subjects: 2, Subscribers: 3, Replicas: 3,
+	})
+
+	// subjectNodes must map each subject to exactly the set of nodes
+	// hosting its publishers, derived independently from the plan.
+	want := make(map[int]map[int]struct{})
+	for i, subject := range pl.pubSubject {
+		if want[subject] == nil {
+			want[subject] = make(map[int]struct{})
+		}
+		want[subject][pl.pubNode[i]] = struct{}{}
+	}
+	require.Equal(t, want, subjectNodes(pl))
+}
+
+func TestReadinessConverged(t *testing.T) {
+	t.Parallel()
+
+	trackers := []*probeTracker{newProbeTracker(), newProbeTracker()}
+	required := []map[int]struct{}{{0: {}}, {0: {}, 1: {}}}
+
+	require.False(t, isReady(trackers, required))
+	trackers[0].observe(0)
+	trackers[1].observe(0)
+	require.False(t, isReady(trackers, required))
+	require.Contains(t, unreadySubscribers(trackers, required), "subscriber 1")
+	trackers[1].observe(1)
+	require.True(t, isReady(trackers, required))
+}

coderd/x/nats/natsbench/report.go

@@ -0,0 +1,234 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"math"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/dustin/go-humanize"
+)
+
+// ScenarioResult pairs a scenario with its run outcome.
+type ScenarioResult struct {
+	Scenario Scenario
+	Result   *Result
+	Err      error
+}
+
+// valid reports whether the run produced trustworthy numbers: it
+// completed without error. Dropped messages do NOT invalidate a run;
+// they are reported in the Drops column alongside the throughput the
+// run did achieve.
+func (r ScenarioResult) valid() bool {
+	return r.Err == nil && r.Result != nil
+}
+
+// RenderMarkdown writes grouped markdown tables, one per payload size
+// in first-seen order. Failed runs (a non-nil error) render INVALID
+// instead of throughput numbers; a Status column appears only for
+// groups that contain a failed run, so clean matrices stay compact.
+// Dropped messages are a normal metric in the Drops column, not a
+// failure.
+func RenderMarkdown(w io.Writer, results []ScenarioResult) error {
+	var b strings.Builder
+	for gi, group := range groupByPayload(results) {
+		if gi > 0 {
+			_, _ = b.WriteString("\n")
+		}
+		_, _ = fmt.Fprintf(&b, "### Payload %s\n\n", formatPayload(group.payload))
+		renderGroup(&b, group.rows)
+	}
+	_, err := io.WriteString(w, b.String())
+	return err
+}
+
+func renderGroup(b *strings.Builder, rows []ScenarioResult) {
+	withStatus := false
+	for _, row := range rows {
+		if !row.valid() {
+			withStatus = true
+			break
+		}
+	}
+
+	// The Status column is left-aligned (free text); the rest are
+	// numeric and right-aligned.
+	headers := []string{"Replicas", "Subjects", "Publishers", "Subscribers", "Messages", "Converge", "Pubs/sec", "Deliveries/sec", "Drops"}
+	aligns := []alignment{alignRight, alignRight, alignRight, alignRight, alignRight, alignRight, alignRight, alignRight, alignRight}
+	if withStatus {
+		headers = append(headers, "Status")
+		aligns = append(aligns, alignLeft)
+	}
+
+	table := [][]string{headers}
+	for _, row := range rows {
+		cells, status := rowCells(row)
+		if withStatus {
+			cells = append(cells, status)
+		}
+		table = append(table, cells)
+	}
+	writeAlignedTable(b, table, aligns)
+}
+
+type alignment int
+
+const (
+	alignLeft alignment = iota
+	alignRight
+)
+
+// writeAlignedTable writes a GitHub-flavored markdown table whose raw
+// text also lines up in a fixed-width terminal: every cell is padded to
+// its column's widest value. table[0] is the header row.
+func writeAlignedTable(b *strings.Builder, table [][]string, aligns []alignment) {
+	widths := make([]int, len(table[0]))
+	for _, rowCells := range table {
+		for col, cell := range rowCells {
+			widths[col] = max(widths[col], len(cell))
+		}
+	}
+
+	writeRow := func(cells []string) {
+		_, _ = b.WriteString("|")
+		for col, cell := range cells {
+			_, _ = fmt.Fprintf(b, " %s |", pad(cell, widths[col], aligns[col]))
+		}
+		_, _ = b.WriteString("\n")
+	}
+
+	writeRow(table[0])
+
+	// Separator row: plain dashes sized to each column.
+	_, _ = b.WriteString("|")
+	for _, width := range widths {
+		_, _ = fmt.Fprintf(b, " %s |", strings.Repeat("-", width))
+	}
+	_, _ = b.WriteString("\n")
+
+	for _, cells := range table[1:] {
+		writeRow(cells)
+	}
+}
+
+// pad widens s to width, on the left for right-aligned columns and on
+// the right otherwise.
+func pad(s string, width int, align alignment) string {
+	gap := width - len(s)
+	if gap <= 0 {
+		return s
+	}
+	if align == alignRight {
+		return strings.Repeat(" ", gap) + s
+	}
+	return s + strings.Repeat(" ", gap)
+}
+
+type payloadGroup struct {
+	payload int
+	rows    []ScenarioResult
+}
+
+func groupByPayload(results []ScenarioResult) []payloadGroup {
+	var groups []payloadGroup
+	index := make(map[int]int)
+	for _, result := range results {
+		size := result.Scenario.Config.PayloadSize
+		gi, ok := index[size]
+		if !ok {
+			gi = len(groups)
+			index[size] = gi
+			groups = append(groups, payloadGroup{payload: size})
+		}
+		groups[gi].rows = append(groups[gi].rows, result)
+	}
+	return groups
+}
+
+// rowCells returns the shape and measured cells for one result, plus a
+// status string. The cells are: replicas, subjects, publishers,
+// subscribers, messages, cluster convergence time, pubs/sec,
+// deliveries/sec, drops. Failed runs render INVALID rates and a status
+// describing why; valid runs render rates and an "ok" status that
+// callers may drop for all-clean groups. A valid run that dropped
+// messages still renders its rates, with the loss in the Drops column.
+func rowCells(row ScenarioResult) (cells []string, status string) {
+	cfg := row.Scenario.Config
+	if row.Result != nil {
+		cfg = row.Result.Config
+	}
+	cells = []string{
+		strconv.Itoa(cfg.Replicas),
+		strconv.Itoa(cfg.Subjects),
+		strconv.Itoa(cfg.Publishers),
+		strconv.Itoa(cfg.Subscribers),
+		humanize.Comma(int64(cfg.Messages)),
+		formatConvergence(row),
+	}
+
+	if row.valid() {
+		return append(cells,
+			formatRate(row.Result.PubsPerSec),
+			formatRate(row.Result.DeliveriesPerSec),
+			formatDrops(row.Result),
+		), "ok"
+	}
+	return append(cells, "INVALID", "INVALID", formatDrops(row.Result)), shortReason(row)
+}
+
+// formatDrops renders the drop count and its percentage of expected
+// deliveries, or "-" when there is no result to measure.
+func formatDrops(res *Result) string {
+	if res == nil {
+		return "-"
+	}
+	if res.Drops == 0 {
+		return "0"
+	}
+	pct := 0.0
+	if res.Expected > 0 {
+		pct = 100 * float64(res.Drops) / float64(res.Expected)
+	}
+	return fmt.Sprintf("%s (%.2f%%)", humanize.Comma(res.Drops), pct)
+}
+
+// shortReason summarizes why a run failed in one table-safe line.
+func shortReason(row ScenarioResult) string {
+	if row.Err == nil {
+		return "no result"
+	}
+	msg := row.Err.Error()
+	if i := strings.IndexByte(msg, '\n'); i >= 0 {
+		msg = msg[:i]
+	}
+	const maxLen = 80
+	if len(msg) > maxLen {
+		msg = msg[:maxLen] + "..."
+	}
+	return strings.ReplaceAll(msg, "|", "\\|")
+}
+
+// formatRate renders a rate as a comma-separated integer.
+func formatRate(rate float64) string {
+	return humanize.Comma(int64(math.Round(rate)))
+}
+
+// formatConvergence renders the cluster convergence time, or "-" for
+// single-node runs and runs without a result, which have no gate.
+func formatConvergence(row ScenarioResult) string {
+	if row.Result == nil || row.Result.Config.Replicas <= 1 {
+		return "-"
+	}
+	return row.Result.ConvergenceDuration.Round(100 * time.Microsecond).String()
+}
+
+// formatPayload renders a payload size as KiB when it divides evenly.
+func formatPayload(size int) string {
+	if size >= 1024 && size%1024 == 0 {
+		return fmt.Sprintf("%d KiB", size/1024)
+	}
+	return fmt.Sprintf("%d B", size)
+}

coderd/x/nats/natsbench/report_internal_test.go

@@ -0,0 +1,168 @@
+package main
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/xerrors"
+
+	"github.com/coder/coder/v2/testutil"
+)
+
+func TestFormatPayload(t *testing.T) {
+	t.Parallel()
+
+	require.Equal(t, "8 KiB", formatPayload(Payload8KB))
+	require.Equal(t, "64 KiB", formatPayload(Payload64KB))
+	require.Equal(t, "100 B", formatPayload(100))
+}
+
+func TestRenderMarkdown(t *testing.T) {
+	t.Parallel()
+
+	valid := ScenarioResult{
+		Scenario: Scenario{
+			Name:   "8KiB-1r",
+			Config: Config{Messages: 100000, PayloadSize: Payload8KB, Replicas: 1},
+		},
+		Result: &Result{
+			Config:           Config{Messages: 100000, PayloadSize: Payload8KB, Replicas: 1},
+			Published:        100000,
+			Delivered:        500000,
+			PublishDuration:  time.Second,
+			DeliverDuration:  2 * time.Second,
+			PubsPerSec:       100000,
+			DeliveriesPerSec: 250000,
+		},
+	}
+	// A run that dropped messages is still valid: it renders real rates
+	// plus the loss in the Drops column.
+	dropped := ScenarioResult{
+		Scenario: Scenario{
+			Name:   "8KiB-5r",
+			Config: Config{Messages: 100000, PayloadSize: Payload8KB, Replicas: 5},
+		},
+		Result: &Result{
+			Config:           Config{Messages: 100000, PayloadSize: Payload8KB, Replicas: 5},
+			Expected:         100000,
+			Published:        100000,
+			Delivered:        97500,
+			Drops:            2500,
+			PublishDuration:  time.Second,
+			DeliverDuration:  time.Second,
+			PubsPerSec:       80000,
+			DeliveriesPerSec: 195000,
+		},
+	}
+	failed := ScenarioResult{
+		Scenario: Scenario{
+			Name:   "64KiB-10r",
+			Config: Config{Messages: 20000, PayloadSize: Payload64KB, Replicas: 10},
+		},
+		Err: xerrors.New("readiness gate: timed out\nsecond line is omitted"),
+	}
+
+	var b strings.Builder
+	require.NoError(t, RenderMarkdown(&b, []ScenarioResult{valid, dropped, failed}))
+	out := b.String()
+
+	require.Contains(t, out, "### Payload 8 KiB")
+	require.Contains(t, out, "### Payload 64 KiB")
+	require.Contains(t, out, "Drops")
+	require.Contains(t, out, "100,000")
+	require.Contains(t, out, "250,000")
+	// The dropped run renders its throughput and its loss percentage,
+	// not INVALID.
+	require.Contains(t, out, "195,000")
+	require.Contains(t, out, "2,500 (2.50%)")
+	// Only the failed run (a non-nil error) renders INVALID and a Status
+	// column.
+	require.Contains(t, out, "Status")
+	require.Contains(t, out, "INVALID")
+	require.Contains(t, out, "readiness gate: timed out")
+	require.NotContains(t, out, "second line is omitted")
+
+	// Every body row has the same width as the header, so columns line
+	// up in a terminal.
+	assertAlignedTable(t, out)
+}
+
+// assertAlignedTable checks that all table rows in a rendered report
+// (lines starting with "|") within each contiguous block share the same
+// rune width.
+func assertAlignedTable(t *testing.T, out string) {
+	t.Helper()
+	width := -1
+	for _, line := range strings.Split(out, "\n") {
+		if !strings.HasPrefix(line, "|") {
+			width = -1
+			continue
+		}
+		if width < 0 {
+			width = len([]rune(line))
+			continue
+		}
+		require.Equal(t, width, len([]rune(line)), "row width mismatch:\n%s", line)
+	}
+}
+
+func TestRenderMarkdownCleanGroupOmitsStatus(t *testing.T) {
+	t.Parallel()
+
+	result := func(replicas int, pubs, dels float64, converge time.Duration) ScenarioResult {
+		cfg := Config{
+			Messages: 100000, PayloadSize: Payload8KB, Replicas: replicas,
+			Subjects: 10, Publishers: 10, Subscribers: 50,
+		}
+		return ScenarioResult{
+			Scenario: Scenario{Config: cfg},
+			Result: &Result{
+				Config: cfg, PubsPerSec: pubs, DeliveriesPerSec: dels,
+				ConvergenceDuration: converge,
+			},
+		}
+	}
+
+	var b strings.Builder
+	require.NoError(t, RenderMarkdown(&b, []ScenarioResult{
+		result(1, 100000, 250000, 0),
+		result(5, 90000, 220000, 25*time.Millisecond),
+	}))
+	out := b.String()
+
+	// The shape and measured columns are reported alongside throughput.
+	for _, header := range []string{"Replicas", "Subjects", "Publishers", "Subscribers", "Messages", "Converge", "Pubs/sec", "Deliveries/sec"} {
+		require.Contains(t, out, header)
+	}
+	// A clean group omits the conditional Status column.
+	require.NotContains(t, out, "Status")
+	require.Contains(t, out, "250,000")
+	require.Contains(t, out, "220,000")
+	// Single-replica rows have no gate; multi-replica rows show the
+	// convergence time.
+	require.Contains(t, out, "25ms")
+	assertAlignedTable(t, out)
+}
+
+func TestDefaultScenarios(t *testing.T) {
+	t.Parallel()
+
+	scenarios := DefaultScenarios()
+	require.Len(t, scenarios, 6)
+	seen := make(map[string]struct{})
+	for _, sc := range scenarios {
+		seen[sc.Name] = struct{}{}
+		cfg := sc.Config
+		cfg.Timeout = testutil.WaitShort
+		require.NoError(t, cfg.validate())
+		if sc.Config.PayloadSize == Payload64KB && sc.Config.Replicas > 1 {
+			require.Less(t, sc.Config.Messages, DefaultMessages,
+				"64KiB cluster runs must reduce the message count")
+		} else {
+			require.Equal(t, DefaultMessages, sc.Config.Messages)
+		}
+	}
+	require.Len(t, seen, 6, "scenario names must be unique")
+}

coderd/x/nats/natsbench/scenario.go

@@ -0,0 +1,76 @@
+package main
+
+import "fmt"
+
+// Scenario names one benchmark configuration in the standard matrix.
+type Scenario struct {
+	Name   string
+	Config Config
+}
+
+// DefaultConns is the default size of the publisher and subscriber
+// connection pools in the standard matrix. Production ships a single
+// connection each, but the benchmarks use three to match the prior
+// natsbench harness and to exercise cross-subject parallelism.
+const DefaultConns = 3
+
+// DefaultSeed seeds the pseudorandom node placement. With the matrix
+// shape (10 publishers) and 10 replicas, most seeds leave several nodes
+// publisher-less (only ~1 in 2750 seeds covers every node), which would
+// skew cross-node routing measurements. 2068 was selected because it
+// spreads publishers perfectly evenly across nodes at every matrix
+// replica count (1, 5, and 10): one publisher per node at 10 replicas,
+// two per node at 5, as TestDefaultSeedSpreadsEvenly verifies. Override
+// with -seed to sample a different, more uneven (and arguably more
+// realistic) placement.
+const DefaultSeed int64 = 2068
+
+// DefaultScenarios returns the standard matrix: payloads of 8 KiB and
+// 64 KiB at 1, 5, and 10 replicas. The 64 KiB cluster runs use a
+// reduced total message count because the fan-out byte volume is
+// memory-heavy.
+// Random node placement (see buildPlan) makes the multi-replica runs
+// exercise cross-node routing.
+//
+// The returned configs leave Timeout (and Seed) unset; callers set them
+// before passing the configs to Run (the CLI does this from its
+// -timeout and -seed flags).
+func DefaultScenarios() []Scenario {
+	const (
+		subjects        = 10
+		publishers      = 10
+		subscribers     = 50
+		reducedMessages = 20000
+	)
+	payloads := []struct {
+		label string
+		size  int
+	}{
+		{"8KiB", Payload8KB},
+		{"64KiB", Payload64KB},
+	}
+
+	var scenarios []Scenario
+	for _, payload := range payloads {
+		for _, replicas := range []int{1, 5, 10} {
+			messages := DefaultMessages
+			if payload.size == Payload64KB && replicas > 1 {
+				messages = reducedMessages
+			}
+			scenarios = append(scenarios, Scenario{
+				Name: fmt.Sprintf("%s-%dr", payload.label, replicas),
+				Config: Config{
+					Messages:       messages,
+					PayloadSize:    payload.size,
+					Subjects:       subjects,
+					Publishers:     publishers,
+					Subscribers:    subscribers,
+					Replicas:       replicas,
+					PublishConns:   DefaultConns,
+					SubscribeConns: DefaultConns,
+				},
+			})
+		}
+	}
+	return scenarios
+}

coderd/x/nats/natsbench/topology.go

@@ -0,0 +1,127 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"slices"
+	"sync"
+	"time"
+
+	natsserver "github.com/nats-io/nats-server/v2/server"
+	"golang.org/x/xerrors"
+
+	"cdr.dev/slog/v3"
+	"github.com/coder/coder/v2/coderd/x/nats"
+)
+
+// topology owns the embedded pubsub nodes for one benchmark run.
+type topology struct {
+	nodes []*nats.Pubsub
+}
+
+// closeAll shuts down every node. Pubsub.Close is idempotent and
+// always returns nil.
+func (t *topology) closeAll() {
+	for _, node := range t.nodes {
+		_ = node.Close()
+	}
+}
+
+// staticPeerFetcher is a mutable nats.PeerFetcher seeded after every
+// node in the cluster has started.
+type staticPeerFetcher struct {
+	mu    sync.Mutex
+	addrs []string
+}
+
+var _ nats.PeerFetcher = (*staticPeerFetcher)(nil)
+
+func (f *staticPeerFetcher) PrimaryPeerAddresses() []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return slices.Clone(f.addrs)
+}
+
+func (f *staticPeerFetcher) set(addrs []string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.addrs = slices.Clone(addrs)
+}
+
+// buildTopology starts cfg.Replicas embedded pubsub nodes. For
+// multi-replica runs it wires a full mesh: every node learns the other
+// nodes' route addresses through its peer fetcher and refreshes its
+// routes. Route convergence is not assumed here; the readiness gate
+// proves it before the measured phase.
+func buildTopology(ctx context.Context, logger slog.Logger, cfg Config) (*topology, error) {
+	// The route auth token is unique per run so concurrent benchmark
+	// processes on one host can never mesh with each other.
+	token := fmt.Sprintf("natsbench-%d", time.Now().UnixNano())
+
+	// One fetcher shared by every node: each node filters its own
+	// address out of the route list, so they can all be handed the
+	// full set of addresses.
+	fetcher := &staticPeerFetcher{}
+	top := &topology{nodes: make([]*nats.Pubsub, 0, cfg.Replicas)}
+	for i := range cfg.Replicas {
+		opts := pubsubOptions(cfg)
+		opts.ClusterAuthToken = token
+		opts.PeerFetcher = fetcher
+		node, err := nats.New(ctx, logger.Named(fmt.Sprintf("node%d", i)), opts)
+		if err != nil {
+			top.closeAll()
+			return nil, xerrors.Errorf("create node %d: %w", i, err)
+		}
+		top.nodes = append(top.nodes, node)
+	}
+
+	if cfg.Replicas > 1 {
+		addrs := make([]string, len(top.nodes))
+		for i, node := range top.nodes {
+			addr, err := routeAddress(node)
+			if err != nil {
+				top.closeAll()
+				return nil, xerrors.Errorf("node %d route address: %w", i, err)
+			}
+			addrs[i] = addr
+		}
+		fetcher.set(addrs)
+		for _, node := range top.nodes {
+			node.RefreshPeers()
+		}
+	}
+	return top, nil
+}
+
+// pubsubOptions maps the benchmark config onto nats.Options. The
+// cluster route listener always uses a random port: a zero ClusterPort
+// means the production default 6222, which both collides across nodes
+// and triggers peer-address port rewriting.
+func pubsubOptions(cfg Config) nats.Options {
+	opts := nats.Options{
+		InProcess:      cfg.InProcess,
+		PublishConns:   cfg.PublishConns,
+		SubscribeConns: cfg.SubscribeConns,
+		ClusterHost:    "127.0.0.1",
+		ClusterPort:    natsserver.RANDOM_PORT,
+	}
+	if cfg.LocalQueueMsgs > 0 {
+		opts.PendingLimits.Msgs = cfg.LocalQueueMsgs
+	}
+	if cfg.LocalQueueBytes > 0 {
+		opts.PendingLimits.Bytes = cfg.LocalQueueBytes
+	}
+	if cfg.MaxPending > 0 {
+		opts.MaxPending = cfg.MaxPending
+	}
+	return opts
+}
+
+// routeAddress returns the node's cluster route address as a NATS URL.
+func routeAddress(node *nats.Pubsub) (string, error) {
+	addr := node.Server.ClusterAddr()
+	if addr == nil {
+		return "", xerrors.New("server has no cluster listener")
+	}
+	return "nats://" + addr.String(), nil
+}

coderd/x/nats/natsbench/workload.go

@@ -0,0 +1,383 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"runtime"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	natsserver "github.com/nats-io/nats-server/v2/server"
+	"golang.org/x/xerrors"
+
+	"cdr.dev/slog/v3"
+	"github.com/coder/coder/v2/coderd/database/pubsub"
+)
+
+// subscriberState tracks one subscriber's exact delivery accounting.
+type subscriberState struct {
+	expect    int
+	delivered atomic.Int64
+	tracker   *probeTracker
+}
+
+// workload wires subscribers and publishers onto a topology and runs
+// the measured phase with exact accounting and bounded waits.
+type workload struct {
+	logger slog.Logger
+	top    *topology
+	pl     plan
+	cfg    Config
+
+	subs    []*subscriberState
+	cancels []func()
+	// dropSignals counts ErrDroppedMessages deliveries. It is a
+	// diagnostic signal only: the authoritative loss count is Expected
+	// minus Delivered, because these signals coalesce and cross-node
+	// routed loss never signals at all.
+	dropSignals atomic.Int64
+
+	published atomic.Int64
+	// outstanding counts subscribers that have not yet reached their
+	// expected delivery count; allDone closes when it hits zero.
+	outstanding atomic.Int64
+	allDone     chan struct{}
+	// convergence is how long the readiness gate took to propagate
+	// interest across the cluster; zero for single-node runs.
+	convergence time.Duration
+	// settleWindow is the delivery quiescence window used when a run
+	// dropped messages and can never reach its exact count. It defaults
+	// to defaultSettleWindow; tests override it to stay fast.
+	settleWindow time.Duration
+}
+
+// runWorkload executes one benchmark run on an already-built topology:
+// subscribe everywhere, prove cluster readiness, release all publishers
+// together, and account for every expected delivery.
+func runWorkload(ctx context.Context, logger slog.Logger, top *topology, pl plan, cfg Config) (*Result, error) {
+	w := &workload{
+		logger:       logger,
+		top:          top,
+		pl:           pl,
+		cfg:          cfg,
+		allDone:      make(chan struct{}),
+		settleWindow: defaultSettleWindow,
+	}
+	defer w.cancelAll()
+	if err := w.subscribe(); err != nil {
+		return nil, err
+	}
+
+	if len(top.nodes) > 1 {
+		trackers := make([]*probeTracker, len(w.subs))
+		for j, st := range w.subs {
+			trackers[j] = st.tracker
+		}
+		convergence, err := awaitTopologyReady(ctx, top, pl, cfg.Timeout, trackers)
+		if err != nil {
+			return nil, xerrors.Errorf("readiness gate: %w", err)
+		}
+		w.convergence = convergence
+		logger.Debug(ctx, "cluster converged", slog.F("duration", convergence))
+	}
+
+	// Both durations are measured from this instant. Goroutine spawn
+	// cost is negligible against the publish phase, so the publishers
+	// start as they are launched rather than behind a barrier.
+	hot := time.Now()
+	pubDone, pubErrCh := w.startPublishers()
+
+	if err := w.awaitPhase(ctx, "publish", pubDone); err != nil {
+		return w.buildResult(time.Since(hot), time.Since(hot)), err
+	}
+	// pubDone closing implies pubErrCh is already closed (both happen
+	// after the publishers' WaitGroup), so this drain terminates.
+	var pubErrs []error
+	for err := range pubErrCh {
+		pubErrs = append(pubErrs, err)
+	}
+	if err := errors.Join(pubErrs...); err != nil {
+		return w.buildResult(time.Since(hot), time.Since(hot)), xerrors.Errorf("publish: %w", err)
+	}
+	for _, idx := range pl.pubNodes {
+		if err := top.nodes[idx].Flush(); err != nil {
+			return w.buildResult(time.Since(hot), time.Since(hot)), xerrors.Errorf("flush publisher node %d: %w", idx, err)
+		}
+	}
+	publishDur := time.Since(hot)
+
+	deliverDur, err := w.awaitDelivery(ctx, hot)
+	if err != nil {
+		return w.buildResult(publishDur, deliverDur), err
+	}
+
+	res := w.buildResult(publishDur, deliverDur)
+	if res.Drops > 0 {
+		logger.Warn(ctx, "run dropped messages",
+			slog.F("expected", res.Expected),
+			slog.F("delivered", res.Delivered),
+			slog.F("drops", res.Drops),
+			slog.F("drop_signals", w.dropSignals.Load()),
+		)
+	}
+	return res, nil
+}
+
+// subscribe registers every subscriber. SubscribeWithErr flushes the
+// SUB on its subscribe connection before returning, so once this
+// returns every subscriber's interest is registered with its local
+// server; cross-node interest propagation is proven separately by the
+// readiness gate.
+func (w *workload) subscribe() error {
+	for j := range w.pl.subSubject {
+		st := &subscriberState{
+			expect:  w.pl.expectPerSub[j],
+			tracker: newProbeTracker(),
+		}
+		w.subs = append(w.subs, st)
+		if st.expect > 0 {
+			w.outstanding.Add(1)
+		}
+		node := w.top.nodes[w.pl.subNode[j]]
+		cancel, err := node.SubscribeWithErr(subjectName(w.pl.subSubject[j]), w.listener(st))
+		if err != nil {
+			return xerrors.Errorf("register subscriber %d: %w", j, err)
+		}
+		w.cancels = append(w.cancels, cancel)
+	}
+	return nil
+}
+
+// listener builds the delivery callback for one subscriber: probes feed
+// the readiness tracker, errors poison the run, and benchmark payloads
+// count toward the exact expected total.
+func (w *workload) listener(st *subscriberState) pubsub.ListenerWithErr {
+	return func(ctx context.Context, message []byte, err error) {
+		if err != nil {
+			if !xerrors.Is(err, pubsub.ErrDroppedMessages) {
+				w.logger.Error(ctx, "unexpected subscriber error", slog.Error(err))
+			}
+			w.dropSignals.Add(1)
+			return
+		}
+		if node, ok := probeNode(message); ok {
+			st.tracker.observe(node)
+			return
+		}
+		// Each subscriber decrements outstanding exactly once, when its
+		// delivered count first equals its expectation, so the atomic
+		// reaches zero exactly once and closes allDone exactly once.
+		if st.delivered.Add(1) == int64(st.expect) {
+			if w.outstanding.Add(-1) == 0 {
+				close(w.allDone)
+			}
+		}
+	}
+}
+
+func (w *workload) cancelAll() {
+	for _, cancel := range w.cancels {
+		cancel()
+	}
+}
+
+// startPublishers launches one goroutine per publisher and returns a
+// channel closed when every publisher finished and a channel of publish
+// errors that is closed at the same time. The error channel is buffered
+// to the publisher count so a failing publisher never blocks, and each
+// publisher sends at most one error.
+func (w *workload) startPublishers() (<-chan struct{}, <-chan error) {
+	payload := make([]byte, w.cfg.PayloadSize)
+	errCh := make(chan error, len(w.pl.perPubMsgs))
+	var wg sync.WaitGroup
+	for i := range w.pl.perPubMsgs {
+		wg.Go(func() {
+			node := w.top.nodes[w.pl.pubNode[i]]
+			subject := subjectName(w.pl.pubSubject[i])
+			for range w.pl.perPubMsgs[i] {
+				if err := node.Publish(subject, payload); err != nil {
+					errCh <- xerrors.Errorf("publisher %d on %s: %w", i, subject, err)
+					return
+				}
+				w.published.Add(1)
+			}
+		})
+	}
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(errCh)
+		close(done)
+	}()
+	return done, errCh
+}
+
+// awaitPhase blocks until the phase signal fires, failing on context
+// cancellation or the per-phase timeout. Timeouts carry full
+// diagnostics so a stuck run is debuggable. Dropped messages do not
+// fail a phase: they are accounted as a metric, not a failure.
+func (w *workload) awaitPhase(ctx context.Context, phase string, signal <-chan struct{}) error {
+	timer := time.NewTimer(w.cfg.Timeout)
+	defer timer.Stop()
+	select {
+	case <-signal:
+		return nil
+	case <-ctx.Done():
+		return xerrors.Errorf("%s phase canceled: %w", phase, ctx.Err())
+	case <-timer.C:
+		return xerrors.Errorf("%s phase timed out after %s:\n%s", phase, w.cfg.Timeout, w.diagnostics())
+	}
+}
+
+// deliveryPollInterval is how often awaitDelivery samples the delivery
+// counter to detect quiescence. It only bounds the precision of the
+// quiescence path; the exact-count fast path fires immediately on
+// allDone regardless of this cadence.
+const deliveryPollInterval = 100 * time.Millisecond
+
+// defaultSettleWindow is how long the delivery counter must stay flat
+// before a run that dropped messages is declared complete. It is a fixed
+// internal constant rather than a knob: only a run that drops messages
+// ever waits it out, and on loopback a backlogged subscriber catches up
+// in milliseconds, so five seconds is ample headroom. Too short would
+// overcount drops; too long only slows runs that already dropped.
+const defaultSettleWindow = 5 * time.Second
+
+// awaitDelivery waits for the deliver phase to finish and returns its
+// duration measured from hot.
+//
+// A zero-drop run reaches its exact expected count, closing allDone, and
+// finishes precisely with no settle delay. A run that dropped messages
+// can never reach that count, so it instead completes by quiescence: the
+// total delivered counter is polled, and once it has not advanced for
+// settleWindow the phase is declared complete. The duration is measured
+// to the last observed progress, not to the end of the settle window, so
+// the idle wait never inflates the delivery rate.
+//
+// The counter is read by summing the per-subscriber delivery atomics, so
+// no global counter or per-delivery timestamp is added to the hot
+// delivery path.
+func (w *workload) awaitDelivery(ctx context.Context, hot time.Time) (time.Duration, error) {
+	timer := time.NewTimer(w.cfg.Timeout)
+	defer timer.Stop()
+	poll := time.NewTicker(deliveryPollInterval)
+	defer poll.Stop()
+
+	lastCount := w.totalDelivered()
+	lastProgress := time.Now()
+	for {
+		select {
+		case <-w.allDone:
+			return time.Since(hot), nil
+		case <-ctx.Done():
+			return time.Since(hot), xerrors.Errorf("deliver phase canceled: %w", ctx.Err())
+		case <-timer.C:
+			return time.Since(hot), xerrors.Errorf("deliver phase timed out after %s:\n%s", w.cfg.Timeout, w.diagnostics())
+		case now := <-poll.C:
+			count := w.totalDelivered()
+			if count != lastCount {
+				lastCount = count
+				lastProgress = now
+				continue
+			}
+			if now.Sub(lastProgress) >= w.settleWindow {
+				return lastProgress.Sub(hot), nil
+			}
+		}
+	}
+}
+
+// totalDelivered sums every subscriber's delivered count. It is called
+// from the quiescence poller, off the hot delivery path.
+func (w *workload) totalDelivered() int64 {
+	var total int64
+	for _, st := range w.subs {
+		total += st.delivered.Load()
+	}
+	return total
+}
+
+// diagnostics renders subscriber shortfalls, per-node server stats, and
+// a goroutine dump for timeout errors.
+func (w *workload) diagnostics() string {
+	var b strings.Builder
+	const maxShortfalls = 20
+	short := 0
+	for j, st := range w.subs {
+		got := st.delivered.Load()
+		if got >= int64(st.expect) {
+			continue
+		}
+		short++
+		if short <= maxShortfalls {
+			_, _ = fmt.Fprintf(&b, "subscriber %d (subject %s, node %d): delivered %d of %d\n",
+				j, subjectName(w.pl.subSubject[j]), w.pl.subNode[j], got, st.expect)
+		}
+	}
+	if short > maxShortfalls {
+		_, _ = fmt.Fprintf(&b, "... and %d more subscribers short\n", short-maxShortfalls)
+	}
+	_, _ = fmt.Fprintf(&b, "published: %d, drop signals: %d\n", w.published.Load(), w.dropSignals.Load())
+
+	for i, node := range w.top.nodes {
+		varz, err := node.Server.Varz(&natsserver.VarzOptions{})
+		if err != nil {
+			_, _ = fmt.Fprintf(&b, "node %d: varz error: %v\n", i, err)
+			continue
+		}
+		_, _ = fmt.Fprintf(&b, "node %d: connections=%d routes=%d subscriptions=%d slow_consumers=%d in_msgs=%d out_msgs=%d\n",
+			i, varz.Connections, varz.Routes, varz.Subscriptions, varz.SlowConsumers, varz.InMsgs, varz.OutMsgs)
+	}
+
+	_, _ = b.WriteString("goroutine dump:\n")
+	_, _ = b.WriteString(goroutineDump())
+	return b.String()
+}
+
+// buildResult snapshots counters into a Result. Drops is the exact
+// shortfall between expected and observed deliveries, the authoritative
+// loss count. Rates are computed for valid and dropping runs alike,
+// since a dropping run still has a meaningful throughput for what it
+// delivered.
+func (w *workload) buildResult(publishDur, deliverDur time.Duration) *Result {
+	delivered := w.totalDelivered()
+	expected := int64(w.pl.totalExpected)
+	res := &Result{
+		Config:              w.cfg,
+		Expected:            expected,
+		Published:           w.published.Load(),
+		Delivered:           delivered,
+		Drops:               max(0, expected-delivered),
+		ConvergenceDuration: w.convergence,
+		PublishDuration:     publishDur,
+		DeliverDuration:     deliverDur,
+		PubsPerSec:          ratePerSec(w.published.Load(), publishDur),
+		DeliveriesPerSec:    ratePerSec(delivered, deliverDur),
+	}
+	return res
+}
+
+// ratePerSec computes count/dur, returning 0 for non-positive
+// durations.
+func ratePerSec(count int64, dur time.Duration) float64 {
+	if dur <= 0 {
+		return 0
+	}
+	return float64(count) / dur.Seconds()
+}
+
+// goroutineDump captures all goroutine stacks, growing the buffer until
+// the dump fits or a hard cap is reached.
+func goroutineDump() string {
+	buf := make([]byte, 1<<20)
+	for {
+		n := runtime.Stack(buf, true)
+		if n < len(buf) || len(buf) >= 16<<20 {
+			return string(buf[:n])
+		}
+		buf = make([]byte, len(buf)*2)
+	}
+}

coderd/x/nats/natsbench/workload_internal_test.go

@@ -0,0 +1,85 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/coder/coder/v2/coderd/database/pubsub"
+	"github.com/coder/coder/v2/testutil"
+)
+
+func TestListenerRecordsDropsAsMetric(t *testing.T) {
+	t.Parallel()
+
+	w := &workload{
+		logger: testutil.Logger(t),
+		cfg:    Config{Timeout: testutil.WaitShort},
+	}
+	st := &subscriberState{expect: 5, tracker: newProbeTracker()}
+
+	// A dropped-message delivery is counted as a signal but does not
+	// advance delivery progress; drops no longer fail the run.
+	w.listener(st)(context.Background(), nil, pubsub.ErrDroppedMessages)
+	require.EqualValues(t, 1, w.dropSignals.Load())
+	require.EqualValues(t, 0, st.delivered.Load())
+}
+
+func TestAwaitDeliveryExactCount(t *testing.T) {
+	t.Parallel()
+
+	ctx := testutil.Context(t, testutil.WaitShort)
+	w := &workload{
+		logger:       testutil.Logger(t),
+		cfg:          Config{Timeout: testutil.WaitShort},
+		allDone:      make(chan struct{}),
+		settleWindow: testutil.WaitShort,
+	}
+	st := &subscriberState{expect: 3, tracker: newProbeTracker()}
+	w.subs = []*subscriberState{st}
+	w.outstanding.Store(1)
+
+	// Deliver exactly the expected count: allDone fires and the deliver
+	// phase finishes precisely, without waiting out the settle window.
+	listener := w.listener(st)
+	go func() {
+		for range 3 {
+			listener(ctx, []byte("x"), nil)
+		}
+	}()
+
+	dur, err := w.awaitDelivery(ctx, time.Now())
+	require.NoError(t, err)
+	require.Positive(t, dur)
+	require.EqualValues(t, 3, w.totalDelivered())
+}
+
+func TestAwaitDeliveryQuiescence(t *testing.T) {
+	t.Parallel()
+
+	ctx := testutil.Context(t, testutil.WaitShort)
+	w := &workload{
+		logger: testutil.Logger(t),
+		cfg:    Config{Timeout: testutil.WaitShort},
+		// A short settle window keeps the test fast; the timeout is far
+		// larger so quiescence, not the hard timeout, ends the phase.
+		allDone:      make(chan struct{}),
+		settleWindow: deliveryPollInterval / 2,
+	}
+	// A subscriber that never reaches its expectation models a dropped
+	// message: allDone can never close, so the phase must complete by
+	// quiescence once the delivery counter stays flat.
+	w.subs = []*subscriberState{{expect: 100}}
+
+	dur, err := w.awaitDelivery(ctx, time.Now())
+	require.NoError(t, err)
+	require.GreaterOrEqual(t, dur, time.Duration(0))
+
+	select {
+	case <-w.allDone:
+		t.Fatal("allDone closed despite a permanent shortfall")
+	default:
+	}
+}

coderd/x/nats/pubsub.go

@@ -17,8 +17,42 @@ import (
 	"github.com/coder/coder/v2/coderd/database/pubsub"
 )
 
-// DefaultMaxPending is the per-client outbound pending byte budget.
-const DefaultMaxPending int64 = 128 << 20
+// DefaultServerMaxPendingBytes caps how many bytes the embedded NATS server will
+// hold in memory for a single client connection while waiting to write
+// them out to that connection. Each message the server needs to deliver
+// to a connection is queued in that connection's outbound buffer until
+// the socket can accept it; if the consumer reads slower than messages
+// arrive, the buffer grows. When it exceeds this cap, NATS declares the
+// connection a slow consumer and drops messages rather than buffer
+// without bound.
+//
+// The connection that fills this buffer in practice is the subscribe
+// connection: the server writes every message bound for a replica's
+// local subscribers out over its subscribe connection pool, which
+// defaults to a single connection. In a cluster, cross-node fan-out
+// therefore concentrates all of a replica's inbound deliveries on that
+// one connection's outbound buffer. Benchmarking high-fanout cluster
+// workloads (10 subjects, 10 publishers, 50 subscribers) showed the 128
+// MiB default overflowing and dropping 10-15% of deliveries, while 256
+// MiB and above dropped none; 512 MiB is chosen for headroom.
+//
+// This is a ceiling, not a reservation: the buffer grows only with
+// actual backlog, so a connection that keeps up holds nearly nothing and
+// raising the cap costs memory only during the overload bursts it
+// absorbs.
+const DefaultServerMaxPendingBytes int64 = 512 << 20
+
+// DefaultClientMaxPendingBytes is the pending byte limit applied via
+// SetPendingLimits to each coalesced *natsgo.Subscription when
+// PendingLimits.Bytes is zero. Unlike DefaultServerMaxPendingBytes,
+// which bounds the server-side outbound buffer for a whole connection,
+// this bounds the nats.go client's per-subscription pending buffer:
+// messages the client has received from the server but the
+// subscription's async message handler has not yet dispatched. When
+// that handler falls behind and the buffer overflows, NATS marks the
+// subscription a slow consumer and drops messages, which the package
+// surfaces as pubsub.ErrDroppedMessages.
+const DefaultClientMaxPendingBytes = 512 * 1024 * 1024
 
 const (
 	defaultClusterName   = "coder"
@@ -31,9 +65,9 @@ var errClosed = xerrors.New("nats pubsub closed")
 // PendingLimits configures per-subscription NATS pending limits set
 // via SetPendingLimits on each *natsgo.Subscription.
 type PendingLimits struct {
-	// Msgs is the per-subscription pending message limit. Positive
-	// values also set each local listener queue capacity.
-	// Zero uses the package default. Negative disables this limit.
+	// Msgs is the per-subscription pending message limit. Zero or
+	// negative disables the message limit, leaving the byte limit
+	// (PendingLimits.Bytes) as the only per-subscription bound.
 	Msgs int
 
 	// Bytes is the per-subscription pending byte limit.
@@ -202,7 +236,7 @@ func defaultPendingLimits(in PendingLimits) PendingLimits {
 		out.Msgs = -1
 	}
 	if out.Bytes == 0 {
-		out.Bytes = 512 * 1024 * 1024
+		out.Bytes = DefaultClientMaxPendingBytes
 	}
 	return out
 }

coderd/x/nats/server.go

@@ -20,7 +20,7 @@ func buildServerOptions(opts Options) (*natsserver.Options, error) {
 	}
 	maxPending := opts.MaxPending
 	if maxPending <= 0 {
-		maxPending = DefaultMaxPending
+		maxPending = DefaultServerMaxPendingBytes
 	}
 
 	sopts := &natsserver.Options{