mesh-host/internal/link/session_test.go

package link

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

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

// quietLink is a link with no broker behind it. A report is refused once the context it is published on is cancelled,
// as the bus's is ("reporting: context canceled"), or while refusing.
type quietLink struct {
	mu       sync.Mutex
	reports  []Report
	refusing bool
	lost     chan error
}

func newQuietLink() *quietLink { return &quietLink{lost: make(chan error, 1)} }

func (l *quietLink) Report(ctx context.Context, _ string, body []byte) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.refusing {
		return errors.New("refused")
	}
	var r Report
	if err := json.Unmarshal(body, &r); err != nil {
		return err
	}
	l.reports = append(l.reports, r)
	return nil
}
func (l *quietLink) Lost() <-chan error { return l.lost }
func (l *quietLink) Close()             {}

func (l *quietLink) said() []Report {
	l.mu.Lock()
	defer l.mu.Unlock()
	return append([]Report(nil), l.reports...)
}

func aMember(t *testing.T) (Membership, ed25519.PrivateKey) {
	t.Helper()
	public, private, err := ed25519.GenerateKey(nil)
	if err != nil {
		t.Fatal(err)
	}
	return Membership{Node: "n1", Signer: public}, private
}

// R6-A AC7: the facts event carries the node-engine's own writer, epoch and sequence (novox/hq ADR 0227 §2, ADR 0539
// §9): one epoch per start, a sequence one higher for every event, so the controller can refuse older facts.
func TestTheFactsEventCarriesTheNodeEnginesWriterEpochAndSequence(t *testing.T) {
	m, _ := aMember(t)
	l := newQuietLink()
	s := &Session{Membership: m, Timeout: time.Second}
	s.Attach(context.Background(), l)

	if !s.SayFacts(context.Background(), Report{Firewall: "none"}) || !s.SayFacts(context.Background(), Report{}) {
		t.Fatal("the facts were not taken")
	}
	got := l.said()
	if len(got) != 2 {
		t.Fatalf("%d facts events were said", len(got))
	}
	for _, r := range got {
		if r.Writer != Writer || r.Node != "n1" || r.Epoch <= 0 {
			t.Fatalf("the facts do not say who wrote them: %+v", r)
		}
	}
	if got[0].Epoch != got[1].Epoch || got[1].Sequence != got[0].Sequence+1 || got[0].Sequence <= 0 {
		t.Fatalf("the order is not one epoch with a rising sequence: %+v then %+v", got[0].Order, got[1].Order)
	}

	// A later start orders after this one with nothing kept on disk.
	time.Sleep(2 * time.Millisecond)
	later := &Session{Membership: m, Timeout: time.Second}
	later.Attach(context.Background(), l)
	later.SayFacts(context.Background(), Report{})
	if last := l.said()[2]; last.Epoch <= got[1].Epoch {
		t.Fatalf("a later start has epoch %d, not above %d", last.Epoch, got[1].Epoch)
	}
}

// Facts made with no link open are not kept, and nothing is said: the next link says them as they are then.
func TestFactsWithNoLinkOpenAreNotSaid(t *testing.T) {
	m, _ := aMember(t)
	l := newQuietLink()
	s := &Session{Membership: m, Timeout: time.Second}
	if s.SayFacts(context.Background(), Report{}) {
		t.Fatal("facts were said with no link")
	}
	s.Attach(context.Background(), l)
	s.Detach(l)
	if s.Linked() || s.SayFacts(context.Background(), Report{}) || len(l.said()) != 0 {
		t.Fatal("facts were said on a link that was let go")
	}
}

// A link opening says the node's facts, off the link's own loop.
func TestALinkOpeningSaysTheFacts(t *testing.T) {
	m, _ := aMember(t)
	l := newQuietLink()
	opened := make(chan struct{}, 1)
	s := &Session{Membership: m, Timeout: time.Second, OnLink: func(context.Context) { opened <- struct{}{} }}
	s.Attach(context.Background(), l)
	select {
	case <-opened:
	case <-time.After(2 * time.Second):
		t.Fatal("a link opening did not ask for the facts")
	}
}

// aBus is a connection to a real server, because some claims here are about one: set MESH_TEST_NATS to a server with
// JetStream, or the test is skipped.
func aBus(t *testing.T) (*nats.Conn, nats.JetStreamContext) {
	t.Helper()
	url := os.Getenv("MESH_TEST_NATS")
	if url == "" {
		t.Skip("MESH_TEST_NATS unset")
	}
	conn, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(conn.Close)
	js, err := conn.JetStream()
	if err != nil {
		t.Fatal(err)
	}
	want := &nats.StreamConfig{Name: "CONTROL", Subjects: []string{"mesh.control.*.report", "mesh.control.enrol"},
		Retention: nats.WorkQueuePolicy}
	if _, err := js.StreamInfo(want.Name); err != nil {
		if _, err := js.AddStream(want); err != nil {
			t.Fatal(err)
		}
	}
	if err := js.PurgeStream(want.Name); err != nil {
		t.Fatal(err)
	}
	return conn, js
}

// eventuallyHeld waits for a condition the bus reaches on its own time.
func eventuallyHeld(t *testing.T, what string, ok func() bool) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for !ok() {
		if time.Now().After(deadline) {
			t.Fatal(what)
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// refusingLink is a quiet link the bus refused subscriptions to.
type refusingLink struct {
	*quietLink
	refused chan string
}

func (l *refusingLink) Refused() <-chan string { return l.refused }

// The sweep of 2026-10-30: every engine was refused its node seat's subscriptions at 12:10, by a user list composed
// without them, and nats asks a refused subscription again only on a new connection; the seat answered nobody all
// afternoon. A refused subscription is read from the server's words, said, and the link opened again after a pause
// that grows while it is refused again; a refused publish is not one.
func TestASubscriptionTheBusRefusedIsAskedAgainOnANewLink(t *testing.T) {
	// The server's words, as the client hands them on (journal of nox-mesh-host on novox, 2026-10-30 12:10:44).
	for _, row := range []struct {
		err  error
		want string
	}{
		{fmt.Errorf("%w: %s", nats.ErrPermissionViolation,
			`Permissions Violation for Subscription to "mesh.seat.node-engine.tool.root-free.novox"`),
			"mesh.seat.node-engine.tool.root-free.novox"},
		{errors.Join(fmt.Errorf("%w: %s", nats.ErrPermissionViolation,
			`Permissions Violation for Subscription to "$SRV.PING.node-engine"`), errors.New("stopping service: x")),
			"$SRV.PING.node-engine"},
		{fmt.Errorf("%w: %s", nats.ErrPermissionViolation,
			`Permissions Violation for Publish to "$JS.API.STREAM.INFO.KV_mesh-host_health"`), ""},
		{errors.New("nats: timeout"), ""},
	} {
		got, ok := RefusedSubscription(row.err)
		if got != row.want || ok != (row.want != "") {
			t.Errorf("%v reads as %q, %v", row.err, got, ok)
		}
		l := &natsLink{refused: make(chan string, 1)}
		l.asyncError(nil, nil, row.err)
		if heard := len(l.refused); heard != map[bool]int{true: 1}[row.want != ""] {
			t.Errorf("%v handed %d refusals on", row.err, heard)
		}
	}

	m, _ := aMember(t)
	q := &Session{Membership: m, Timeout: time.Second, RefusedPause: 50 * time.Millisecond}
	var said []string
	var mu sync.Mutex
	say := func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }
	for _, want := range []time.Duration{50 * time.Millisecond, 100 * time.Millisecond} {
		l := &refusingLink{quietLink: newQuietLink(), refused: make(chan string, 2)}
		l.refused <- "mesh.seat.node-engine.tool.root-free.n1"
		l.refused <- "$SRV.PING"
		began := time.Now()
		done := make(chan error, 1)
		go func() { done <- serve(context.Background(), l, m, q, say, time.Second) }()
		select {
		case err := <-done:
			if err == nil || !strings.Contains(err.Error(), "refused 2 of this node's subscriptions, "+
				"mesh.seat.node-engine.tool.root-free.n1 first") {
				t.Fatalf("the link ended with %v", err)
			}
			if took := time.Since(began); took < want {
				t.Errorf("the link was dropped after %s, before its pause of %s", took, want)
			}
		case <-time.After(5 * time.Second):
			t.Fatal("a link the bus refused a subscription was kept, and the subscription never asked again")
		}
	}
	mu.Lock()
	defer mu.Unlock()
	if n := strings.Count(strings.Join(said, "\n"), "the bus refused this node a subscription to "+
		"mesh.seat.node-engine.tool.root-free.n1"); n != 2 {
		t.Errorf("the refusal was said %d times in two links: %q", n, said)
	}
	if q.refusalsHeld(); q.refusedPause() != 50*time.Millisecond {
		t.Error("a link that held its subscriptions did not start from the first pause again")
	}
}