package main import ( "context" "encoding/json" "os/exec" "path/filepath" "testing" "time" "github.com/nats-io/nats-server/v2/server" "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/testbus" ) // novox/hq issue 327, replayed with only what the controller had before its fix, so it can be laid over // the older commit. On 2026-10-08 the bus's connection total rose by 5.2 a minute while the same 16 // connections stayed open, and three calls of `conditions` in one second added three: each verb ran as a // process of the controller's own binary, which dialled the bus, logged in and left. The operator's // channel reads `conditions` at least once a minute. A verb that only reads, served by the serving // controller, opens no connection of its own. func TestReplay327(t *testing.T) { bus := testbus.Start(t) serving, err := broker.Dial(bus.ClientURL()) if err != nil { t.Fatal(err) } defer serving.Close() if err := serving.EnsureControllerBuckets(); err != nil { t.Fatal(err) } keeper, err := keeperOn(context.Background(), serving.Conn()) if err != nil { t.Fatal(err) } // The serving controller: its keeper and its connection, as serve sets them. keptBefore, connBefore := conditionsFrom, handActConn conditionsFrom, handActConn = keeper, serving.Conn() defer func() { conditionsFrom, handActConn = keptBefore, connBefore }() // A verb that runs as a process of its own runs this controller's binary, on this bus. asAProcess(t, bus.ClientURL()) handlers, _, err := seatToolHandlers() if err != nil { t.Fatal(err) } accepted := func() uint64 { v, err := bus.Varz(nil) if err != nil { t.Fatal(err) } return v.TotalConnections } before := accepted() for i := 0; i < 3; i++ { answer, err := handlers["conditions"](context.Background(), json.RawMessage(`{}`)) if err != nil { t.Fatal(err) } if a, ok := answer.(verbAnswer); !ok || !a.OK { t.Fatalf("conditions answered %+v", answer) } } if opened := accepted() - before; opened != 0 { t.Fatalf("three conditions calls opened %d connection(s) to the bus; the serving controller is on it already", opened) } } // asAProcess makes a verb that runs as a process of its own run this package's binary, on the bus at url: // built into the test's own directory, which goes with the test. func asAProcess(t *testing.T, url string) { t.Helper() path := filepath.Join(t.TempDir(), "mesh-controller") if out, err := exec.Command("go", "build", "-o", path, ".").CombinedOutput(); err != nil { t.Fatalf("the controller could not be built to run a verb as its own process: %v: %s", err, out) } was := ownImage ownImage = func() string { return path } t.Cleanup(func() { ownImage = was }) t.Setenv(broker.NATSVar, url) t.Setenv(broker.CertificateVar, "") } // closedNames are the names of the connections the bus saw closed. func closedNames(t *testing.T, bus *server.Server) []string { t.Helper() deadline := time.Now().Add(5 * time.Second) var names []string for time.Now().Before(deadline) { connz, err := bus.Connz(&server.ConnzOptions{State: server.ConnClosed}) if err != nil { t.Fatal(err) } names = names[:0] for _, c := range connz.Conns { names = append(names, c.Name) } if len(names) > 0 { return names } time.Sleep(50 * time.Millisecond) } return names }