mesh-controller/internal/broker/controller_buckets_test.go

package broker

import (
	"context"
	"github.com/nats-io/nats.go"
	"slices"
	"strings"
	"testing"
	"time"

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

	"github.com/novox/mesh-controller/internal/testbus"
)

// **The controller may write every bucket it writes** (novox/hq to-be 45 §1, issue 269). Writing a
// key is a publish to the bucket's own subject, which the management interface's grant does not
// cover: the cancelled sets' writes timed out for want of this, and the controller's own buckets
// would have.
func TestTheControllerMayWriteEveryBucketItWrites(t *testing.T) {
	p, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}
	want := []string{"$KV." + CallsBucket + ".>", "$KV." + HandActsBucket + ".>",
		"$KV." + AssignmentsBucket + ".>", "$KV." + DeclaredConfigBucket + ".>",
		"$KV." + PredefinedBucket + ".>"}
	for _, seat := range seatsTheControllerAsks {
		if hasCancelledSet(seat) {
			want = append(want, "$KV."+CancelledSetName(seat)+".>")
		}
	}
	for _, subject := range want {
		if !slices.Contains(p.Publish, subject) {
			t.Errorf("the controller may not publish %s, so it cannot write that bucket", subject)
		}
	}
	if slices.Contains(p.Publish, "$KV.>") {
		t.Error("the controller may write any bucket, a module's state included")
	}
}

// **A module reads a record mirror through `reads`, and never writes it** (ADR 0201 §1, design 54 §5,
// design 57 F3): granted the direct reads and watches of the bucket, and refused at composition a grant to
// write it, whose writer is the controller.
func TestAModuleReadsTheControllersMirrorsAndWritesNone(t *testing.T) {
	reader := Principal{Kind: KindModule, Node: "one", Module: "mesh-config", PasswordHash: "x",
		Reads: []string{"mesh-controller." + MirrorAssignments, "mesh-controller." + MirrorDeclaredConfig,
			"mesh-controller." + MirrorPredefined}}
	p, err := PermissionsFor(reader)
	if err != nil {
		t.Fatal(err)
	}
	for _, bucket := range []string{AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket} {
		for _, subject := range []string{"$JS.API.STREAM.INFO.KV_" + bucket, "$JS.API.DIRECT.GET.KV_" + bucket + ".>",
			"$JS.API.CONSUMER.CREATE.KV_" + bucket + ".>"} {
			if !slices.Contains(p.Publish, subject) {
				t.Errorf("a module reading %s may not publish %s", bucket, subject)
			}
		}
		for _, subject := range p.Publish {
			if SubjectsOverlap(subject, "$KV."+bucket+".>") {
				t.Errorf("a module reading %s may write it through %s", bucket, subject)
			}
		}
		if err := CheckWriters(reader, []string{"$KV." + bucket + ".>"}); err == nil {
			t.Errorf("a module granted a write of %s would compose", bucket)
		}
	}
}

// **A machine's node tools may say they are there, as that machine and no other** (novox/hq to-be 45
// S11), and the controller may hear the bus's advisories and ask who answers — read-only, named.
func TestTheWatchedSignalsMayBeSaidAndHeard(t *testing.T) {
	tools, err := PermissionsFor(Principal{Kind: KindNodeTools, Node: "anchor", Module: RuntimeModule, PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(tools.Publish, "mesh.control.anchor.tools-alive") {
		t.Error("the node tools may not say they are there")
	}
	for _, s := range tools.Publish {
		if strings.Contains(s, "tools-alive") && s != "mesh.control.anchor.tools-alive" {
			t.Errorf("the node tools may say %s", s)
		}
	}
	controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range BusAdvisories {
		if !slices.Contains(controller.Subscribe, s) {
			t.Errorf("the controller may not hear %s", s)
		}
	}
	if !slices.Contains(controller.Publish, "$SRV.INFO") || slices.Contains(controller.Subscribe, "$JS.EVENT.>") {
		t.Error("the controller may not ask who answers, or hears every API call")
	}
}

// The controller's record of what it asked the operator is bounded (correctness review of 2026-10-08): one
// value a key, a month's age, and a size it cannot outgrow.
func TestWhatTheControllerAskedIsBounded(t *testing.T) {
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	kv, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	bucket, err := kv.KeyValue(ctx, AskedBucket)
	if err != nil {
		t.Fatal(err)
	}
	status, err := bucket.Status(ctx)
	if err != nil {
		t.Fatal(err)
	}
	info := status.(*jetstream.KeyValueBucketStatus).StreamInfo()
	if status.History() != 1 || status.TTL() != AskedKeptFor || info.Config.MaxBytes <= 0 || info.Config.MaxBytes > 64<<20 {
		t.Errorf("history %d, age %s, bytes %d", status.History(), status.TTL(), info.Config.MaxBytes)
	}
}

// **A listed mirror is no undeclared bucket** (ADR 0356 §3, ADR 0201 §16): the controller's list of its mirrors
// stands for their declaration, so the report of state nothing declares never names one.
func TestAListedMirrorIsNotReportedAsUndeclared(t *testing.T) {
	names := []string{"gone_state"}
	for _, m := range ControllerMirrors() {
		names = append(names, BucketName(ControllerSeat, m))
	}
	undeclared, err := RaiseBuckets(fakeBuckets{names: names}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Equal(undeclared, []string{"gone_state"}) {
		t.Fatalf("the report of undeclared state names %v", undeclared)
	}
}

// **The mirrors' list and the writers table agree** (ADR 0356 §3, to-be 45 §1): every listed mirror is a row of
// the table whose one writer is the controller, on its bucket alone; and every row that writes a bucket of the
// controller's named as a mirror is on the list.
func TestTheMirrorsListAndTheWritersTableAgree(t *testing.T) {
	controller := Principal{Kind: KindController}
	module := Principal{Kind: KindModule, Module: ControllerSeat, Node: "one"}
	rowed := map[string]bool{}
	for _, row := range WritersTable {
		if !strings.HasSuffix(row.State, " mirror") {
			continue
		}
		for _, subject := range row.Subjects {
			bucket := strings.TrimSuffix(strings.TrimPrefix(subject, "$KV."), ".>")
			rowed[bucket] = true
			if row.Writes == nil || !row.Writes(controller, subject) || row.Writes(module, subject) {
				t.Errorf("%q is not written by the controller alone", row.State)
			}
		}
	}
	listed := map[string]bool{}
	for _, m := range ControllerMirrors() {
		bucket := BucketName(ControllerSeat, m)
		listed[bucket] = true
		if !rowed[bucket] {
			t.Errorf("the mirror %s is listed and no row of the writers table writes it", bucket)
		}
		if !IsControllerBucket(bucket) || !slices.Contains(ControllerBuckets(), bucket) {
			t.Errorf("the mirror %s is listed and not asserted as the controller's own", bucket)
		}
	}
	for bucket := range rowed {
		if !listed[bucket] {
			t.Errorf("the writers table writes the mirror %s, which the list does not name", bucket)
		}
	}
}

// **The controller may ask the vault's intake verbs in its own name** (design 57's V2, issue 675): the vault
// takes a value only from the controller, which it knows from the subject that names its caller — so the
// controller is granted that subject for the verbs the vault's holder serves, and no more of the vault's seat.
// `get` among them since issue 755 (ADR 0349 §16): the vault answers the controller a core credential sealed to its
// machine alone, which the controller carries and cannot open.
// Composed from the records, as the issued tools are: the installer's first user list carries none.
func TestTheControllerMayAskTheVaultInItsOwnName(t *testing.T) {
	vault := Declared{Module: "mesh-vault", Holds: []Seat{{Name: "mesh-vault",
		Serves: []string{"get", "accept", "secret-ask", "issued", "rotate"}}}}
	users, err := Users(Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {vault}}})
	if err != nil {
		t.Fatal(err)
	}
	controller := users[0]
	if strings.Join(controller.AsksTheVault, ",") != "accept,get,issued,rotate,secret-ask" {
		t.Fatalf("composed %v", controller.AsksTheVault)
	}
	p, err := PermissionsFor(controller)
	if err != nil {
		t.Fatal(err)
	}
	for _, verb := range []string{"accept", "secret-ask", "issued", "rotate", "get"} {
		if !MayPublish(p, CalledSubject("mesh.seat.mesh-vault.tool."+verb, ControllerName)) {
			t.Errorf("the controller may not ask the vault's %s in its own name", verb)
		}
		if MayPublish(p, CalledSubject("mesh.seat.mesh-vault.tool."+verb, "anchor.node-tools")) {
			t.Errorf("the controller may ask the vault's %s in another's name", verb)
		}
	}
	if MayPublish(p, CalledSubject("mesh.seat.mesh-vault.tool.move-in", ControllerName)) {
		t.Error("the controller may move a value in: that is a node-engine's alone")
	}
	// A vault that serves none of them yet composes no grant, nor does genesis.
	older := Declared{Module: "mesh-vault", Holds: []Seat{{Name: "mesh-vault", Serves: []string{"move-in"}}}}
	users, _ = Users(Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {older}}})
	if len(users[0].AsksTheVault) != 0 {
		t.Errorf("a vault serving none composed %v", users[0].AsksTheVault)
	}
	genesis, _ := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
	for _, s := range genesis.Publish {
		if strings.Contains(s, "mesh-vault") {
			t.Errorf("the installer's controller may publish %s", s)
		}
	}
}

// **The controller may ask mesh-config's `config` in its own name** (issue 925): move-config and the settings verb
// set a trusted value through mesh-config, which takes it as the controller carries it only from the subject that
// names the controller as caller. Granted when the records hold a holder serving the verb, and not otherwise;
// the installer's first user list carries none.
func TestTheControllerMayAskMeshConfigInItsOwnName(t *testing.T) {
	config := Declared{Module: "mesh-config", Holds: []Seat{{Name: "mesh-config", Serves: []string{"config"}}}}
	users, err := Users(Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {config}}})
	if err != nil {
		t.Fatal(err)
	}
	if strings.Join(users[0].AsksTheConfig, ",") != "config" {
		t.Fatalf("composed %v", users[0].AsksTheConfig)
	}
	p, err := PermissionsFor(users[0])
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(p.Publish, "mesh.seat.mesh-config.call.config.controller") {
		t.Errorf("the controller may not publish mesh.seat.mesh-config.call.config.controller: %v", p.Publish)
	}
	if !MayPublish(p, CalledSubject("mesh.seat.mesh-config.tool.config", ControllerName)) {
		t.Error("the controller may not ask mesh-config's config in its own name")
	}
	if MayPublish(p, CalledSubject("mesh.seat.mesh-config.tool.config", "anchor.node-tools")) {
		t.Error("the controller may ask mesh-config's config in another's name")
	}
	// No holder serving it composes no grant, nor does genesis.
	other := Declared{Module: "mesh-config", Holds: []Seat{{Name: "mesh-config", Serves: []string{"list"}}}}
	for _, r := range []Records{{Nodes: []string{"novox"}}, {Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {other}}}} {
		users, _ = Users(r)
		if len(users[0].AsksTheConfig) != 0 {
			t.Errorf("composed %v without a holder serving config", users[0].AsksTheConfig)
		}
	}
	genesis, _ := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
	for _, s := range genesis.Publish {
		if strings.Contains(s, "mesh-config") {
			t.Errorf("the installer's controller may publish %s", s)
		}
	}
}

// **The merge window is a bucket whose keys expire** (novox/hq ADR 0407 decision 1, issue 973): a key made with a
// time to live goes when it is up, and a watcher on the bucket hears it go. The controller may write it, and it is
// no undeclared state.
func TestTheMergeWindowsKeysExpireAndAWatcherHearsIt(t *testing.T) {
	controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"})
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(controller.Publish, "$KV."+MergeWindowBucket+".>") {
		t.Error("the controller may not write the merge window's bucket")
	}
	if !IsControllerBucket(MergeWindowBucket) || !slices.Contains(ControllerBuckets(), MergeWindowBucket) {
		t.Error("the merge window's bucket is not asserted as the controller's own")
	}
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	api, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	kv, err := api.KeyValue(ctx, MergeWindowBucket)
	if err != nil {
		t.Fatal(err)
	}
	w, err := kv.WatchAll(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer w.Stop()
	if _, err := kv.Create(ctx, "plan-1.quiet", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil {
		t.Fatal(err)
	}
	for {
		select {
		case e := <-w.Updates():
			if e != nil && e.Key() == "plan-1.quiet" && e.Operation() != jetstream.KeyValuePut {
				return
			}
		case <-ctx.Done():
			t.Fatal("no watcher heard the key expire")
		}
	}
}

// **The merges have a consumer of their own, which a reset of the events consumer leaves alone** (issue 973, novox/hq
// ADR 0407 decision 3): healer H4 re-makes the events consumer from now, dropping what it holds pending; a merge
// waiting in the merges consumer is not that consumer's, and nothing would catch it up.
func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testing.T) {
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	if err := AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	if err := AssertMeshConsumers(js); err != nil {
		t.Fatal(err)
	}
	for _, c := range MeshConsumers() {
		if c.Stream == EventsStream && c.Name == MergesConsumerName {
			if c.Resettable != "" || c.MaxDeliver != 0 || len(c.Filters) != 1 || c.Filters[0] != ControllerFollows[3] {
				t.Errorf("the merges consumer is %+v; want one filter, no max-deliver, not resettable", c)
			}
		}
		if c.Stream == EventsStream && c.Name == ControllerName {
			for _, f := range c.Filters {
				if f == ControllerFollows[3] {
					t.Errorf("the events consumer still follows the merges, which the merges consumer takes")
				}
			}
		}
	}
	if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil {
		t.Fatal(err)
	}
	if _, err := js.Context().Publish(ControllerFollows[0], []byte(`{}`)); err != nil {
		t.Fatal(err)
	}
	merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
	if err != nil || merges.NumPending != 1 {
		t.Fatalf("the merges consumer holds %v (%v); want the merge", merges, err)
	}
	if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil {
		t.Fatal(err)
	}
	merges, err = js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
	if err != nil || merges.NumPending != 1 || merges.Delivered.Stream != 0 {
		t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err)
	}
}

// **Delivering this change loses no merge** (review of issue 973): a merge published under the old filter of the
// `controller` consumer and never acknowledged by it is delivered by the merges consumer once the assertion has
// made it, because it starts where `controller` has acknowledged up to, not from now.
func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(t *testing.T) {
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	if err := AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	old := &nats.ConsumerConfig{Durable: ControllerName, AckPolicy: nats.AckExplicitPolicy, AckWait: 30 * time.Second,
		MaxDeliver: 5, MaxAckPending: 1, DeliverSubject: "_DELIVER.controller.EVENTS", DeliverPolicy: nats.DeliverNewPolicy,
		FilterSubjects: append(append([]string(nil), ControllerFollows...), AgainFilter(ControllerName))}
	if _, err := js.Context().AddConsumer(EventsStream, old); err != nil {
		t.Fatal(err)
	}
	if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil {
		t.Fatal(err)
	}
	if err := AssertMeshConsumers(js); err != nil {
		t.Fatal(err)
	}
	merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
	if err != nil || merges.NumPending != 1 {
		t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err)
	}
}