mesh-catalog/modules/nats/cmd/nats-tools/server_test.go

package main

import (
	"encoding/json"
	"fmt"
	"io"
	"math/rand"
	"net/http"
	"os"
	"regexp"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
)

// The server these tests run is the server the mesh runs. The manifest's `server` artifact is the upstream
// image, written `name:tag@sha256:<digest>` (novox/hq ADR 0480 clause 3): the digest names the image and the
// tag says the release it is. This keeps the tag and the server compiled in here equal: a test of the server
// below is only worth something about the server the mesh actually runs.
func TestTheImageIsTheServerTestedHere(t *testing.T) {
	raw, err := os.ReadFile("../../module.json")
	if err != nil {
		t.Fatal(err)
	}
	var m struct {
		Build struct {
			Artifacts []struct {
				Name string `json:"name"`
				Kind string `json:"kind"`
				From string `json:"from"`
			} `json:"artifacts"`
		} `json:"build"`
	}
	if err := json.Unmarshal(raw, &m); err != nil {
		t.Fatal(err)
	}
	from := ""
	for _, a := range m.Build.Artifacts {
		if a.Name == "server" {
			if a.Kind != "upstream" {
				t.Fatalf("the server artifact is a %q, not an upstream image", a.Kind)
			}
			from = a.From
		}
	}
	said := regexp.MustCompile(`^nats:(\d+\.\d+\.\d+)-alpine@sha256:[0-9a-f]{64}$`).FindStringSubmatch(from)
	if said == nil {
		t.Fatalf("the server artifact is %q: it must read nats:X.Y.Z-alpine@sha256:<digest>, the tag saying the release", from)
	}
	// The server's own word for its release, compiled into it: a test binary's build information may list
	// no dependencies (the build seat's did not, so this read "these tests run no nats-server").
	if server.VERSION != said[1] {
		t.Fatalf("the image runs nats %s and these tests run %s: move them together", said[1], server.VERSION)
	}
}

// **A consumer with several filters is handed every message** (novox/hq issue 266).
//
// The controller follows the forge's merges, build outcomes and providers' standings on one durable
// consumer with seven filters, one message at a time. On nats 2.10.29 such a consumer was moved past
// a message now and then without handing it over: nothing pending, nothing redelivered, the message
// on the stream and its consumer never told. On 2026-10-06 that message was a merge, and the modules
// built from that repository were left behind with nobody told. This is that consumer, under traffic
// shaped like the mesh's — a stream mostly of build logs on ever-new subjects, the followed events
// few among them — and it fails on 2.10.29 (a few percent of the followed events never arrive) and
// passes on the release the image pins.
func TestAConsumerWithSeveralFiltersIsHandedEveryMessage(t *testing.T) {
	if testing.Short() {
		t.Skip("runs a server under load for seconds")
	}
	s, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(5 * time.Second) {
		t.Fatal("the server did not start")
	}
	defer s.Shutdown()

	admin, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	defer admin.Close()
	js, _ := admin.JetStream()
	// The events stream and the controller's consumer on it, as the mesh declares them.
	if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"},
		MaxAge: 168 * time.Hour, MaxMsgsPerSubject: 10000, Storage: nats.FileStorage}); err != nil {
		t.Fatal(err)
	}
	if _, err := js.AddConsumer("EVENTS", &nats.ConsumerConfig{Durable: "controller", DeliverSubject: "_DELIVER.controller.EVENTS",
		FilterSubjects: []string{
			"mesh.mod.mesh-catalog.event.upgraded",
			"mesh.mod.mesh-catalog.event.catching-up",
			"mesh.seat.node-build-agent.event.built",
			"mesh.mod.gitea.event.pull.merged",
			"mesh.seat.mesh-build-machine.event.built",
			"mesh.mod.*.event.provisioner.failing",
			"mesh.mod.*.event.provisioner.recovered",
		},
		AckPolicy: nats.AckExplicitPolicy, AckWait: 30 * time.Second, MaxDeliver: 5, MaxAckPending: 1,
		DeliverPolicy: nats.DeliverNewPolicy}); err != nil {
		t.Fatal(err)
	}

	controller, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	defer controller.Close()
	cjs, _ := controller.JetStream()
	handed := make(chan *nats.Msg, 64)
	if _, err := cjs.ChanSubscribe("", handed, nats.Bind("EVENTS", "controller")); err != nil {
		t.Fatal(err)
	}
	var got sync.Map
	go func() {
		for m := range handed {
			if meta, err := m.Metadata(); err == nil {
				got.Store(meta.Sequence.Stream, true)
			}
			time.Sleep(time.Duration(rand.Intn(20)) * time.Millisecond)
			_ = m.Ack()
		}
	}()

	followed := []string{"mesh.mod.gitea.event.pull.merged", "mesh.seat.node-build-agent.event.built",
		"mesh.mod.postgres.event.provisioner.recovered", "mesh.mod.mesh-catalog.event.catching-up"}
	var mu sync.Mutex
	var sent []uint64
	stop := time.Now().Add(5 * time.Second)
	var wg sync.WaitGroup
	for w := 0; w < 4; w++ {
		wg.Add(1)
		go func(w int) {
			defer wg.Done()
			nc, err := nats.Connect(s.ClientURL())
			if err != nil {
				t.Error(err)
				return
			}
			defer nc.Close()
			pjs, _ := nc.JetStream()
			for i := 0; time.Now().Before(stop); i++ {
				if rand.Intn(40) == 0 {
					if ack, err := pjs.Publish(followed[rand.Intn(len(followed))], []byte(`{}`)); err == nil {
						mu.Lock()
						sent = append(sent, ack.Sequence)
						mu.Unlock()
					}
				} else {
					_, _ = pjs.Publish(fmt.Sprintf("mesh.seat.node-build-agent.event.log.build-%d-%d", w, i/50), make([]byte, 200))
				}
				if rand.Intn(100) == 0 {
					time.Sleep(time.Duration(rand.Intn(300)) * time.Millisecond)
				}
			}
		}(w)
	}
	wg.Wait()

	// Every followed event handed over, given the consumer time to finish.
	deadline := time.Now().Add(20 * time.Second)
	for {
		var missing []uint64
		for _, seq := range sent {
			if _, ok := got.Load(seq); !ok {
				missing = append(missing, seq)
			}
		}
		if len(missing) == 0 {
			return
		}
		info, err := js.ConsumerInfo("EVENTS", "controller")
		if time.Now().After(deadline) || (err == nil && info.NumPending == 0 && info.NumAckPending == 0) {
			t.Fatalf("%d of %d followed events were never handed to the consumer (first at stream sequence %d), "+
				"and it has nothing pending: the server moved past them", len(missing), len(sent), missing[0])
		}
		time.Sleep(200 * time.Millisecond)
	}
}

// The record of closed connections is the server's own, read through its monitoring endpoint as the tool
// reads it (novox/hq issue 327): two connections a client opened and closed are both there, by user and
// name, closed by their client.
func TestTheServersRecordOfClosedConnectionsIsRead(t *testing.T) {
	s, err := server.NewServer(&server.Options{Port: -1, HTTPPort: -1, NoLog: true, NoSigs: true,
		Users: []*server.User{{Username: "controller", Password: "x"}}})
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(10 * time.Second) {
		t.Fatal("no server")
	}
	defer s.Shutdown()
	for i := 0; i < 2; i++ {
		nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("controller", "x"), nats.Name("mesh-controller verb conditions"))
		if err != nil {
			t.Fatal(err)
		}
		nc.Close()
	}
	monitor := func(path string) []byte {
		resp, err := http.Get("http://" + s.MonitorAddr().String() + path)
		if err != nil {
			t.Fatal(err)
		}
		defer resp.Body.Close()
		body, _ := io.ReadAll(resp.Body)
		return body
	}
	var users []ClosedUser
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		out, err := ClosedConnections(monitor("/connz?state=closed&auth=1&limit=10000"), time.Now(), 1, "")
		if err != nil {
			t.Fatal(err)
		}
		if users = out["users"].([]ClosedUser); len(users) == 1 && users[0].Closed == 2 {
			break
		}
		time.Sleep(50 * time.Millisecond)
	}
	if len(users) != 1 || users[0].User != "controller" || users[0].Closed != 2 || users[0].Dropped != 0 ||
		users[0].Names[0].Name != "mesh-controller verb conditions" {
		t.Fatalf("%+v", users)
	}
}