mesh-controller/cmd/mesh-controller/composed_start_test.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"testing"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/compose"
	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go/jetstream"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/inventory"
)

// fakeBuckets is the bus's buckets by name, for deleting.
type fakeBuckets map[string]bool

func (f fakeBuckets) DeleteKeyValue(_ context.Context, bucket string) error {
	if !f[bucket] {
		return jetstream.ErrBucketNotFound
	}
	delete(f, bucket)
	return nil
}

// 1945/AC14 (unit half): the first start deletes the merge window's, the silent sends' and the building plans' buckets
// and says each once; a later start finds none and says nothing; a bucket of anything else is left alone.
func TestTheFirstStartDeletesThePlansBucketsAndSaysEachOnce(t *testing.T) {
	bus := fakeBuckets{"mesh-controller_merge-window": true, "mesh-controller_rollout-silent": true,
		"mesh-controller_plans-building": true, "mesh-controller_grants": true}
	var said []string
	deleted, err := deletePlanStateBuckets(context.Background(), bus, func(b string) { said = append(said, b) })
	if err != nil || len(deleted) != 3 || strings.Join(said, " ") != strings.Join(broker.PlanStateBuckets, " ") {
		t.Fatalf("first start: deleted %v, said %v, %v", deleted, said, err)
	}
	if !bus["mesh-controller_grants"] || len(bus) != 1 {
		t.Errorf("the bus holds %v after the first start", bus)
	}
	said = nil
	if deleted, err := deletePlanStateBuckets(context.Background(), bus, func(b string) { said = append(said, b) }); err != nil || len(deleted)+len(said) != 0 {
		t.Errorf("a later start deleted %v and said %v (%v)", deleted, said, err)
	}
}

// 1945/AC6 (unit half): the controller this one replaced is composed on its node from its own build's manifest at its
// own fingerprint: the composition holds what that manifest declares, not what the controller registered now declares.
func TestTheReplacedControllerIsComposedFromItsOwnManifest(t *testing.T) {
	now := catalogue.Manifest{Module: controllerModule, Version: "6", Resources: []map[string]any{
		{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "six\n"}}}
	before := catalogue.Manifest{Module: controllerModule, Version: "5", Resources: []map[string]any{
		{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "five\n"}}}
	in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now},
		Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true}}}
	fp := sdkdelivery.SourcePrefix2 + strings.Repeat("5", 64)
	c, err := replacedComposition(in, replacedController{node: "anchor", fp: fp, manifest: before})
	if err != nil {
		t.Fatal(err)
	}
	if c.Module != controllerModule || c.FP != fp || c.Node != "anchor" || len(c.Resources) != 1 {
		t.Fatalf("composed %s at %s on %s with %d resources", c.Module, c.FP, c.Node, len(c.Resources))
	}
	var f struct{ Content string }
	_ = json.Unmarshal(c.Resources[0], &f)
	if f.Content != "five\n" {
		t.Errorf("the replaced controller's composition holds %q, its own manifest says five", f.Content)
	}
	if in.Catalogue[controllerModule].Version != "6" {
		t.Errorf("composing the replaced controller changed the Input's catalogue")
	}
}

// 1945/AC6 (unit half): the controller replaced is the newest earlier build of the controller that worked and names a
// fingerprint of the new form, not the one registered now.
func TestTheReplacedControllerIsItsNewestEarlierBuild(t *testing.T) {
	now, before, older := sdkdelivery.SourcePrefix2+strings.Repeat("6", 64), sdkdelivery.SourcePrefix2+strings.Repeat("5", 64),
		sdkdelivery.SourcePrefix2+strings.Repeat("4", 64)
	m := []byte(`{"module":"mesh-controller","version":"5"}`)
	builds := []inventory.Build{
		{ID: "c6", SourceFingerprint: now, Manifest: m},
		{ID: "c5-failed", SourceFingerprint: before, Failed: "no", Manifest: m},
		{ID: "c5", SourceFingerprint: before, Manifest: m},
		{ID: "c4", SourceFingerprint: older, Manifest: m},
	}
	if b, ok := replacedBuild(builds, now); !ok || b.ID != "c5" {
		t.Errorf("the replaced build is %+v (%v), want c5", b, ok)
	}
	if _, ok := replacedBuild(builds[:1], now); ok {
		t.Errorf("a controller with no earlier build replaced one")
	}
}

// 1945/AC14 (unit half): the declarations' stream stays while one node's have names another node-engine than the one
// registered, or none, and goes once every node's names it.
func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) {
	n6 := sdkdelivery.SourcePrefix2 + strings.Repeat("6", 64)
	nodes := []string{"ace", "g14", "novox"}
	if declarationsDone(nodes, map[string]string{"ace": n6, "novox": n6}, n6) {
		t.Errorf("done while g14 says nothing (it sleeps)")
	}
	if declarationsDone(nodes, map[string]string{"ace": n6, "g14": "src2:old", "novox": n6}, n6) {
		t.Errorf("done while g14 runs the old node-engine")
	}
	if !declarationsDone(nodes, map[string]string{"ace": n6, "g14": n6, "novox": n6}, n6) {
		t.Errorf("not done with every node on the new node-engine")
	}
	if declarationsDone(nodes, map[string]string{"ace": "", "g14": "", "novox": ""}, "") || declarationsDone(nil, nil, n6) {
		t.Errorf("done with no fingerprint registered, or no node")
	}
}

// streamsGone is a bus that holds none of the streams asked for; every other part of it is unread.
type streamsGone struct{ jetstream.JetStream }

func (streamsGone) Stream(context.Context, string) (jetstream.Stream, error) {
	return nil, jetstream.ErrStreamNotFound
}

// Issue 1945 (the alarm of a stream the start deleted): once the declarations' stream is gone from the bus, the
// declarations are retired whatever the nodes run now, so no raise, health check, probe or healer expects the stream
// or a node's consumer on it again. A node-engine newer than the registered one, or a node asleep on an older have,
// made the done test false after the stream went, and the bus's objects named it missing again (2026-10-25 14:52).
func TestTheDeclarationsStayRetiredOnceTheirStreamIsGone(t *testing.T) {
	retired, err := func() (retired bool, err error) {
		defer func() {
			if p := recover(); p != nil {
				err = fmt.Errorf("it read what the nodes run although the stream is gone: %v", p)
			}
		}()
		return declarationsRetiredOn(t.Context(), nil, streamsGone{})
	}()
	if err != nil || !retired {
		t.Fatalf("with no declarations' stream on the bus: retired %v (%v), want retired", retired, err)
	}
}