mesh-controller/cmd/mesh-controller/composed_integration_test.go

//go:build integration

package main

import (
	"context"
	"errors"
	"sync"
	"testing"

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

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

// A mesh that composes (novox/hq ADR 0539): the real store (theCatalogue: sudo from the catalogue, mesh-host from the
// node-engine's repository, the private network on anchor and laptop), the controller's signing key in its own
// identity store, and a real bus server this process composes onto, its lease taken by the act as a command's is. Every
// subject published on the bus is heard, so a test can say nothing went to a node's declaration subject.
type composingMesh struct {
	open  *stores
	js    jetstream.JetStream
	conn  *nats.Conn
	mu    sync.Mutex
	heard []string
}

func aComposingMesh(t *testing.T) *composingMesh {
	t.Helper()
	open := theCatalogue(t)
	ident := identity.ForTest(t)
	if _, err := ident.Establish(t.Context()); err != nil {
		t.Fatal(err)
	}
	s := testbus.Start(t)
	t.Setenv(broker.NATSVar, s.ClientURL())
	was := theLease
	theLease = &actor{}
	t.Cleanup(func() { theLease.release(); theLease = was })
	conn, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(conn.Close)
	js, err := jetstream.New(conn)
	if err != nil {
		t.Fatal(err)
	}
	m := &composingMesh{open: open, js: js, conn: conn}
	sub, err := conn.Subscribe(">", func(msg *nats.Msg) {
		m.mu.Lock()
		m.heard = append(m.heard, msg.Subject)
		m.mu.Unlock()
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
	legacy, err := conn.JetStream()
	if err != nil {
		t.Fatal(err)
	}
	// The controller's seat events, kept as the mesh's events stream keeps them.
	if _, err := legacy.AddStream(&nats.StreamConfig{Name: "CONTROLLER_EVENTS", Subjects: []string{"mesh.seat.mesh-controller.event.>"}}); err != nil {
		t.Fatal(err)
	}
	wasEvents := checkEvents
	checkEvents = link.OverNATS{Conn: conn, JS: legacy}
	t.Cleanup(func() { checkEvents = wasEvents })
	withTrunkList(t, true)
	return m
}

// trunkKeys are a done key and the indexer's trunk value naming it.
type trunkKeys struct {
	keyBuckets
	trunk sdkdelivery.Module
}

func (k trunkKeys) ModuleAt(context.Context, string, string) (sdkdelivery.Module, error) {
	return k.trunk, nil
}

// 1945/AC14: the declarations' stream stays while one node's have names another node-engine than the one registered,
// and goes, said once, when every node's names it (ADR 0539 clause 11).
func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
	m := aComposingMesh(t)
	ctx := t.Context()
	legacy, err := m.conn.JetStream()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := legacy.AddStream(&nats.StreamConfig{Name: broker.DeclarationsStream, Subjects: []string{"mesh.node.*.declare"},
		MaxMsgsPerSubject: 1}); err != nil {
		t.Fatal(err)
	}
	// The new node-engine registered at its fingerprint, from its done key.
	n6 := fpOf('6')
	keys, trunk := aKeyFor("mesh-host", "6", n6, "novox/mesh-host", "", "f00d0123456789ab")
	settled, _ := liveTakeInSteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}, m.js).batch(ctx,
		[]delivered{{module: "mesh-host", again: true}}, map[string]string{})
	if len(settled) != 1 || settled[0].err != nil || settled[0].verdict != acknowledged {
		t.Fatalf("the take-in of mesh-host: %+v", settled)
	}
	if _, err := m.js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket, History: sdkdelivery.HaveHistory}); err != nil {
		t.Fatal(err)
	}
	haves, err := m.js.KeyValue(ctx, sdkdelivery.HaveBucket)
	if err != nil {
		t.Fatal(err)
	}
	have := func(node, fp string) {
		key, _ := sdkdelivery.HaveKey(node, "mesh-host")
		raw, _ := sdkdelivery.EncodeHave(sdkdelivery.Have{FP: fp, Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthOK})
		if _, err := haves.Put(ctx, key, raw); err != nil {
			t.Fatal(err)
		}
	}
	have("anchor", n6)
	have("laptop", fpOf('5'))
	if done, err := retireDeclarations(ctx, m.open); err != nil || done {
		t.Fatalf("with laptop on the old node-engine: done %v (%v)", done, err)
	}
	if _, err := m.js.Stream(ctx, broker.DeclarationsStream); err != nil {
		t.Fatalf("the stream went while laptop runs the old node-engine: %v", err)
	}
	have("laptop", n6)
	if done, err := retireDeclarations(ctx, m.open); err != nil || !done {
		t.Fatalf("with every node on the new node-engine: done %v (%v)", done, err)
	}
	if _, err := m.js.Stream(ctx, broker.DeclarationsStream); !errors.Is(err, jetstream.ErrStreamNotFound) {
		t.Errorf("the stream is still there: %v", err)
	}
	if done, err := retireDeclarations(ctx, m.open); err != nil || !done {
		t.Errorf("a second look: done %v (%v)", done, err)
	}
	if !declarationsRetiredNow(ctx, m.open.inventory) {
		t.Errorf("a raise after the retirement would assert the stream again")
	}

	// Issue 1945: a node-engine other than the registered one landing on laptop (a newer build on its way) leaves the
	// declarations retired, so what the health check, the probes and healer H3 expect of the bus names neither the
	// stream nor a node's consumer on it, and asserting the bus's objects again does not make the stream anew.
	have("laptop", fpOf('7'))
	if !declarationsRetiredNow(ctx, m.open.inventory) {
		t.Fatalf("with laptop on another node-engine after the stream went, the declarations are no longer retired")
	}
	streams, consumers, err := expectedBusObjects(ctx, m.open.inventory)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range streams {
		if s.Name == broker.DeclarationsStream {
			t.Errorf("the bus's health check and probe D7 still expect the stream %s", s.Name)
		}
	}
	for _, c := range consumers {
		if c.Stream == broker.DeclarationsStream {
			t.Errorf("the bus's health check and probe D6 still expect %s on %s", c.Name, c.Stream)
		}
	}
	bus, err := broker.Dial(m.conn.ConnectedUrl())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(bus.Close)
	_ = assertOnSend(ctx, m.open.inventory, bus, "  ")
	if _, err := m.js.Stream(ctx, broker.DeclarationsStream); !errors.Is(err, jetstream.ErrStreamNotFound) {
		t.Errorf("healer H3's assertion of the bus's objects made the stream anew: %v", err)
	}
}