mesh-host/internal/link/session_test.go

package link

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"os"
	"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)
	}
}