mesh-host/internal/link/handover_test.go

package link

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

	stdio "git.novox.be/novox/mesh-sdk/go"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/micro"
)

// The hand-over's ask and answer hold these field names; the controller's side holds the same list
// (mesh-controller internal/link, TestTheHandOverAskAndAnswerKeepTheirFieldNames).
func TestTheHandOverAskAndAnswerKeepTheirFieldNames(t *testing.T) {
	body, _ := json.Marshal(HandOverAsk{Node: "laptop", Path: "/srv/notes", By: "jo", Expires: time.Now(), Nonce: "n"})
	if got := keysOf(t, body); got != "by expires node nonce path" {
		t.Fatalf("the ask's fields are %q", got)
	}
	body, _ = json.Marshal(SignedHandOver{Ask: []byte("{}"), Signature: []byte("s")})
	if got := keysOf(t, body); got != "ask signature" {
		t.Fatalf("the signed ask's fields are %q", got)
	}
	body, _ = json.Marshal(HandOverAnswer{Said: "s", Refused: "r"})
	if got := keysOf(t, body); got != "refused said" {
		t.Fatalf("the answer's fields are %q", got)
	}
	if AskHandOverSubject("laptop") != "mesh.node.laptop.ask.hand-over" {
		t.Fatalf("the subject is %q", AskHandOverSubject("laptop"))
	}
	if HandOverContext != "novox-mesh hand-over v1\n" {
		t.Fatalf("the signing context is %q; the controller holds the same words", HandOverContext)
	}
}

func keysOf(t *testing.T, body []byte) string {
	t.Helper()
	var m map[string]any
	if err := json.Unmarshal(body, &m); err != nil {
		t.Fatal(err)
	}
	var keys []string
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return strings.Join(keys, " ")
}

const aNonce = "00112233445566778899aabbccddeeff"

// signedAsk is an ask as the controller makes it, signed with key over the context and its bytes.
func signedAsk(t *testing.T, key ed25519.PrivateKey, ask HandOverAsk) []byte {
	t.Helper()
	raw, err := json.Marshal(ask)
	if err != nil {
		t.Fatal(err)
	}
	body, err := json.Marshal(SignedHandOver{Ask: raw, Signature: ed25519.Sign(key, append([]byte(HandOverContext), raw...))})
	if err != nil {
		t.Fatal(err)
	}
	return body
}

// anAsk is a good ask at now.
func anAsk(now time.Time) HandOverAsk {
	return HandOverAsk{Node: "n1", Path: "/srv/notes", By: " jo through mesh-cli on anchor ", Expires: now.Add(HandOverGood),
		Nonce: aNonce}
}

// **An ask is believed only signed by the mesh, for this node, unexpired, once** (review of issue 356): a
// responder can reach the subject through a reply, so the subject proves nothing. Each refusal below hands
// the handler nothing; the good ask reaches it with its path and who asked, and the same ask again is refused.
func TestAHandOverAskIsBelievedOnlySignedForThisNodeUnexpiredAndOnce(t *testing.T) {
	m, key := aMember(t)
	_, stranger, _ := ed25519.GenerateKey(nil)
	now := time.Date(2026, 10, 9, 18, 0, 0, 0, time.UTC)
	var given []string
	handle := func(path, by string) (string, error) {
		given = append(given, path+" by "+by)
		if path == "/refused" {
			return "", errors.New("not a directory this machine uses as found; nothing was handed over")
		}
		return path + " is handed to the mesh", nil
	}
	check := HandOverCheck{Signer: m.Signer, Node: m.Node, Now: func() time.Time { return now }, Nonces: &Nonces{}}
	with := func(change func(*HandOverAsk)) HandOverAsk {
		a := anAsk(now)
		change(&a)
		return a
	}
	good, _ := json.Marshal(anAsk(now))
	unsignedBody, _ := json.Marshal(SignedHandOver{Ask: good})
	// Signed as a declaration is, over the bytes alone: not a hand-over's signature.
	asDeclaration, _ := json.Marshal(SignedHandOver{Ask: good, Signature: ed25519.Sign(key, good)})
	refusals := map[string][]byte{
		"too large":               signedAsk(t, key, with(func(a *HandOverAsk) { a.Path = "/" + strings.Repeat("a", MaxHandOverAsk) })),
		"not a signed hand-over":  []byte(`not json`),
		"unsigned":                unsignedBody,
		"signed by another key":   signedAsk(t, stranger, anAsk(now)),
		"signed as a declaration": asDeclaration,
		"the bare ask":            good,
		"for another node":        signedAsk(t, key, with(func(a *HandOverAsk) { a.Node = "n2" })),
		"expired":                 signedAsk(t, key, with(func(a *HandOverAsk) { a.Expires = now })),
		"no expiry":               signedAsk(t, key, with(func(a *HandOverAsk) { a.Expires = time.Time{} })),
		"good for too long":       signedAsk(t, key, with(func(a *HandOverAsk) { a.Expires = now.Add(HandOverGood + HandOverSkew + time.Second) })),
		"no nonce":                signedAsk(t, key, with(func(a *HandOverAsk) { a.Nonce = "" })),
		"a short nonce":           signedAsk(t, key, with(func(a *HandOverAsk) { a.Nonce = "0011" })),
		"nobody":                  signedAsk(t, key, with(func(a *HandOverAsk) { a.By = "  " })),
		"two lines":               signedAsk(t, key, with(func(a *HandOverAsk) { a.By = "jo\nroot" })),
		"a name too long":         signedAsk(t, key, with(func(a *HandOverAsk) { a.By = strings.Repeat("x", MaxHandOverBy+1) })),
		"a NUL in the name":       signedAsk(t, key, with(func(a *HandOverAsk) { a.By = "jo\x00" })),
		"no directory":            signedAsk(t, key, with(func(a *HandOverAsk) { a.Path = "" })),
	}
	for name, body := range refusals {
		a := AnswerHandOver(check, handle, body)
		if a.Refused == "" || a.Said != "" || !strings.Contains(a.Refused, "nothing was handed over") {
			t.Fatalf("%s was answered %+v; want a refusal saying nothing was handed over", name, a)
		}
	}
	if len(given) != 0 {
		t.Fatalf("the handler was given %v although every ask was refused", given)
	}
	if a := AnswerHandOver(HandOverCheck{Node: "n1", Nonces: &Nonces{}}, handle, signedAsk(t, key, anAsk(time.Now()))); a.Refused == "" {
		t.Fatal("an engine with no key believed an ask")
	}
	if a := AnswerHandOver(check, nil, signedAsk(t, key, anAsk(now))); a.Refused == "" {
		t.Fatal("an engine with no handler took a hand-over")
	}

	body := signedAsk(t, key, anAsk(now))
	a := AnswerHandOver(check, handle, body)
	if a.Said != "/srv/notes is handed to the mesh" || a.Refused != "" {
		t.Fatalf("the good ask was answered %+v", a)
	}
	if again := AnswerHandOver(check, handle, body); !strings.Contains(again.Refused, "taken already") {
		t.Fatalf("the same ask again was answered %+v", again)
	}
	refused := signedAsk(t, key, with(func(a *HandOverAsk) { a.Path = "/refused"; a.Nonce = strings.Repeat("ab", 16) }))
	if a := AnswerHandOver(check, handle, refused); !strings.HasPrefix(a.Refused, "not a directory") {
		t.Fatalf("the handler's refusal was answered %+v", a)
	}
	if strings.Join(given, "; ") != "/srv/notes by jo through mesh-cli on anchor; /refused by jo through mesh-cli on anchor" {
		t.Fatalf("the handler was given %v", given)
	}
	// Issue 1126: a link opened on the mesh has no signer (the controller that signed these asks ended), so an ask signed
	// by the membership's key, the controller's, is refused and nothing is handed over.
	before := len(given)
	opened := (&natsLink{node: m.Node}).check()
	opened.Now, opened.Nonces = func() time.Time { return now }, &Nonces{}
	late := signedAsk(t, key, with(func(a *HandOverAsk) { a.Nonce = strings.Repeat("cd", 16) }))
	if a := AnswerHandOver(opened, handle, late); !strings.Contains(a.Refused, "not signed by the mesh") || len(given) != before {
		t.Fatalf("a link opened on the mesh answered an ask signed by the controller's key: %+v", a)
	}
	// A nonce is forgotten once its ask has expired, so the set does not grow for ever.
	now = now.Add(HandOverGood + time.Second)
	check.Nonces.take("ff", now.Add(time.Second), now)
	if len(check.Nonces.seen) != 1 {
		t.Fatalf("expired nonces are kept: %v", check.Nonces.seen)
	}
}

// **A request with no reply is nobody's question**: nothing is judged, recorded or said. One with a reply is
// answered with the judgement.
func TestAHandOverIsAnsweredOnlyWhenItCarriesAReply(t *testing.T) {
	m, key := aMember(t)
	now := time.Now()
	called := 0
	handle := func(string, string) (string, error) { called++; return "handed", nil }
	check := HandOverCheck{Signer: m.Signer, Node: m.Node, Nonces: &Nonces{}}
	if body, answer := HandOverReply(check, handle, "", signedAsk(t, key, anAsk(now))); answer || body != nil || called != 0 {
		t.Fatalf("a request with no reply was answered %q %v, handled %d", body, answer, called)
	}
	body, answer := HandOverReply(check, handle, "_INBOX.x", signedAsk(t, key, anAsk(now)))
	var said HandOverAnswer
	if !answer || json.Unmarshal(body, &said) != nil || said.Said != "handed" || called != 1 {
		t.Fatalf("answered %q %v, handled %d", body, answer, called)
	}
}

// handingLink is a quiet link that hears hand-overs, keeping the handler it was given.
type handingLink struct {
	*quietLink
	given chan HandOverFunc
}

func (l handingLink) HearHandOvers(h HandOverFunc) { l.given <- h }

// **The link is given the queue's hand-over** when it is served: what the controller asks reaches the engine's
// judgement, and nothing else.
func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
	m, _ := aMember(t)
	l := handingLink{quietLink: newQuietLink(), given: make(chan HandOverFunc, 1)}
	q := &Session{Membership: m, Timeout: time.Second}
	q.HandOver = func(path, by string) (string, error) { return "the queue's: " + path + " by " + by, nil }
	ctx, stop := context.WithCancel(context.Background())
	served := make(chan error, 1)
	go func() { served <- serve(ctx, l, m, q, nil, time.Second) }()
	select {
	case h := <-l.given:
		if said, err := h("/srv/notes", "jo"); err != nil || said != "the queue's: /srv/notes by jo" {
			t.Fatalf("the link was given another handler: %q %v", said, err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the link was never given the queue's hand-over")
	}
	stop()
	<-served
	// 774 (ADR 0563 §1): a call of the node seat reaches its verb with its caller as the subject proves it, unproven on
	// the subject that names none; a verb the seat does not serve is refused, and the answer is a tool's envelope.
	t.Run("a call of the node seat reaches its verb with its proven caller", func(t *testing.T) {
		h := &HandActs{Node: "n1", State: &MemState{},
			RootFree:         func() RootVerdict { return RootVerdict{Machine: "n1", Free: true, Why: "judged"} },
			JudgeDirectories: func(by string) (string, error) { return "judged for " + by, nil },
			Composition: func(module, digest string, diff bool) (any, error) {
				return map[string]any{"module": module, "digest": digest, "diff": diff}, nil
			},
			Compose: func(module string) (any, error) { return map[string]any{"composed": true, "retried": module}, nil },
			// ADR 0565 §6, §15: the node's key of the engine's state packages.
			Packages: func(write bool) (any, error) {
				key := []byte(`{"node":"n1","packages":[{"package":"curl","modules":["backup","web"],"absent":false}]}`)
				if !write {
					return PackagesAnswer("n1", key, nil), nil
				}
				v := PackagesAnswer("n1", nil, nil)
				wrote := true
				v.Wrote = &wrote
				return v, nil
			}}
		for _, row := range []struct{ verb, subject, want string }{
			{"judge-directories", "mesh.seat.node-engine.call.judge-directories.n1.shanks~node-tools", `"said":"judged for shanks.node-tools"`},
			{"judge-directories", EngineToolSubject("judge-directories", "n1"), `"said":"judged for unproven"`},
			{"judge-directories", "mesh.seat.node-engine.call.judge-directories.n2.shanks~node-tools", `"said":"judged for unproven"`},
			{"root-free", "mesh.seat.node-engine.call.root-free.n1.ace~messenger", `"free":true`},
			// The read verb composition (issue 774) takes its module, digest and diff, as a word or a boolean.
			{"composition", EngineToolSubject("composition", "n1") + " " + `{"module":"postgres","digest":"sha256:a","diff":true}`, `"diff":true,"digest":"sha256:a","module":"postgres"`},
			{"composition", EngineToolSubject("composition", "n1") + " " + `{"module":"postgres","diff":"true"}`, `"diff":true,"digest":"","module":"postgres"`},
			// The recovery verb compose (issue 774) makes a pass now, with a module's apply tried again when named.
			{"compose", EngineToolSubject("compose", "n1"), `"composed":true,"retried":""`},
			{"compose", EngineToolSubject("compose", "n1") + " " + `{"module":"route-proxy"}`, `"retried":"route-proxy"`},
			// The read verb packages (ADR 0565 §6) names its source; with write, what it wrote, and a key the engine has not
			// written is said and not an error.
			{"packages", EngineToolSubject("packages", "n1"), `"source":"mesh-host.packages","key":{"node":"n1"`},
			{"packages", EngineToolSubject("packages", "n1") + " " + `{"write":true}`, `"key":null,"wrote":true`},
			{"packages", EngineToolSubject("packages", "n1") + " " + `{"write":"true"}`, `"note":"the engine has written no key for this node yet`},
			{"unserved", "mesh.seat.node-engine.call.unserved.n1.x", `"error":"the node seat node-engine serves no verb \"unserved\""`},
		} {
			subject, body, ok := strings.Cut(row.subject, " ")
			if !ok {
				body = `{}`
			}
			if got := string(AnswerEngineCall(h.Call, row.verb, "n1", subject, []byte(body))); !strings.Contains(got, row.want) ||
				!strings.Contains(got, `"node":"n1"`) {
				t.Errorf("%s on %s: %s", row.verb, row.subject, got)
			}
		}
		acts, _ := h.Acts(0)
		if got := acts.(map[string]any)["acts"].([]HandAct); len(got) != 3 || got[0].Caller != "unproven" ||
			got[2].Caller != "shanks.node-tools" || got[2].Act != ActJudgeDirectories {
			t.Errorf("the judgements are recorded as %+v", got)
		}
	})
	// Issue 1126 (review of 2026-11-03): two asks of compose-candidate at once, handed over one at a time as an
	// endpoint's subscription hands them, are both answered while the other runs: neither waits for the other, so each
	// bound counts from its own arrival. A verb answered on the subscription answers in order, as before.
	t.Run("two asks of compose-candidate at once are answered side by side", func(t *testing.T) {
		var mu sync.Mutex
		arrived, both := 0, make(chan struct{})
		handle := func(c EngineCall) (any, error) {
			mu.Lock()
			if arrived++; arrived == 2 {
				close(both)
			}
			mu.Unlock()
			select {
			case <-both:
				return "together", nil
			case <-time.After(2 * time.Second): // only when the asks are answered one after the other
				return "alone", nil
			}
		}
		answered := make(chan string, 2)
		answer := EngineAnswer(handle, ActComposeCandidate, "n1", make(chan struct{}, 4))
		for range 2 {
			answer(fakeRequest{subject: EngineToolSubject(ActComposeCandidate, "n1"), answered: answered})
		}
		for range 2 {
			if got := <-answered; !strings.Contains(got, `"result":"together"`) {
				t.Errorf("an ask answered %s; want both answered side by side", got)
			}
		}
		full := make(chan struct{}, 1)
		full <- struct{}{}
		EngineAnswer(handle, ActComposeCandidate, "n1", full)(fakeRequest{subject: "s", answered: answered})
		if got := <-answered; !strings.Contains(got, "asks of compose-candidate at once") {
			t.Errorf("an ask past the bound answered %s", got)
		}
	})
}

// fakeRequest is a seat call as the subscription hands it over, its answer sent to answered.
type fakeRequest struct {
	subject  string
	answered chan string
}

func (r fakeRequest) Respond(b []byte, _ ...micro.RespondOpt) error {
	r.answered <- string(b)
	return nil
}
func (r fakeRequest) RespondJSON(any, ...micro.RespondOpt) error { return nil }
func (r fakeRequest) Error(string, string, []byte, ...micro.RespondOpt) error {
	return nil
}
func (r fakeRequest) Data() []byte           { return []byte(`{}`) }
func (r fakeRequest) Headers() micro.Headers { return nil }
func (r fakeRequest) Subject() string        { return r.subject }
func (r fakeRequest) Reply() string          { return "_INBOX.x" }

// **Against a real server**: a signed request on the node's own subject is answered; one on another node's is
// not heard; a message with no reply is answered by nobody.
func TestNatsAHandOverIsAskedOfTheNode(t *testing.T) {
	conn, js := aBus(t)
	m, key := aMember(t)
	l := &natsLink{conn: conn, js: js, node: m.Node, signer: m.Signer}
	handled := make(chan string, 4)
	l.HearHandOvers(func(path, by string) (string, error) { handled <- path; return "handed " + path, nil })
	defer l.Close()
	if err := conn.Flush(); err != nil {
		t.Fatal(err)
	}
	ask := anAsk(time.Now())
	ask.Nonce = strings.Repeat("cd", 16)
	reply, err := conn.Request(AskHandOverSubject(m.Node), signedAsk(t, key, ask), 5*time.Second)
	if err != nil {
		t.Fatal(err)
	}
	var said HandOverAnswer
	if err := json.Unmarshal(reply.Data, &said); err != nil || said.Said != "handed /srv/notes" {
		t.Fatalf("answered %s, %v", reply.Data, err)
	}
	ask.Nonce = strings.Repeat("ef", 16)
	if err := conn.Publish(AskHandOverSubject(m.Node), signedAsk(t, key, ask)); err != nil {
		t.Fatal(err)
	}
	if _, err := conn.Request(AskHandOverSubject("another"), signedAsk(t, key, ask), 300*time.Millisecond); err == nil {
		t.Fatal("another node's ask was answered here")
	}
	if err := conn.Flush(); err != nil {
		t.Fatal(err)
	}
	time.Sleep(100 * time.Millisecond)
	if len(handled) != 1 {
		t.Fatalf("handled %d asks; want only the one with a reply", len(handled))
	}
	// 774 (ADR 0563 §2, §3): the engine's acts are its node's keys of its declared state on the bus, each written by
	// compare-and-set: a second create and a stale revision lose, and another node's keys are not this node's.
	t.Run("the acts are kept in the bus's own storage under this node's keys", func(t *testing.T) {
		_ = js.DeleteKeyValue(ActsBucket)
		if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: ActsBucket}); err != nil {
			t.Fatal(err)
		}
		q := &Session{Membership: m, Timeout: time.Second}
		q.bus = l
		mine, theirs := BusState{Session: q, Node: m.Node}, BusState{Session: q, Node: "another"}
		rev, err := mine.Create("act.a-1", map[string]string{"state": "asked"})
		if err != nil {
			t.Fatal(err)
		}
		if _, err := mine.Create("act.a-1", map[string]string{}); !errors.Is(err, stdio.ErrChanged) {
			t.Errorf("a second create: %v", err)
		}
		if _, err := mine.Update("act.a-1", map[string]string{"state": "done"}, rev+7); !errors.Is(err, stdio.ErrChanged) {
			t.Errorf("an update at a stale revision: %v", err)
		}
		if _, err := mine.Update("act.a-1", map[string]string{"state": "done"}, rev); err != nil {
			t.Errorf("an update at its revision: %v", err)
		}
		_, _ = theirs.Create("act.b-1", map[string]string{})
		if keys, err := mine.Keys("act."); err != nil || len(keys) != 1 || keys[0] != "act.a-1" {
			t.Errorf("this node's acts are %v, %v", keys, err)
		}
		if e, _ := js.KeyValue(ActsBucket); e != nil {
			if got, err := e.Get(m.Node + ".act.a-1"); err != nil || !strings.Contains(string(got.Value()), "done") {
				t.Errorf("the bucket holds %v, %v", got, err)
			}
		}
		if err := mine.Delete("act.a-1"); err != nil {
			t.Fatal(err)
		}
		if e, err := mine.Get("act.a-1"); err != nil || e != nil {
			t.Errorf("a deleted act reads %v, %v", e, err)
		}
	})
	// 774 (ADR 0563 §1): the node seat is served on this node's subjects alone, naming its caller, and said to discovery.
	t.Run("the node seat is served on this node and discovered", func(t *testing.T) {
		l.ServeEngineSeat(func(c EngineCall) (any, error) { return c.Verb + " for " + c.Caller, nil })
		reply, err := conn.Request("mesh.seat.node-engine.call.acts."+m.Node+".shanks~node-tools", []byte(`{}`), 5*time.Second)
		if err != nil || !strings.Contains(string(reply.Data), `"result":"acts for shanks.node-tools"`) {
			t.Fatalf("the call answered %v, %v", reply, err)
		}
		if _, err := conn.Request(EngineToolSubject("acts", "another"), []byte(`{}`), 300*time.Millisecond); err == nil {
			t.Fatal("another node's verb was answered here")
		}
		// Announced in the shape the console reads a seat's verb in (sweep 2026-10-30): without kind and scope it listed
		// none of it, and <node>/node-engine.root-free answered that nothing in the mesh is called node-engine.
		info, err := conn.Request("$SRV.INFO."+EngineSeat, nil, 5*time.Second)
		if err != nil {
			t.Fatal(err)
		}
		var said struct {
			Endpoints []struct {
				Subject  string            `json:"subject"`
				Metadata map[string]string `json:"metadata"`
			} `json:"endpoints"`
		}
		if err := json.Unmarshal(info.Data, &said); err != nil {
			t.Fatal(err)
		}
		announced := 0
		for _, e := range said.Endpoints {
			md := e.Metadata
			if md["kind"] != "seat" || md["scope"] != "node" || md["seat"] != EngineSeat || md["module"] != "mesh-host" ||
				md["node"] != m.Node || md["description"] == "" || !strings.HasPrefix(md["schema"], "{") {
				t.Errorf("the endpoint on %s announces %v", e.Subject, md)
			}
			if e.Subject == EngineToolSubject("root-free", m.Node) {
				announced++
			}
		}
		if announced != 1 {
			t.Errorf("root-free is announced on its tool subject %d times", announced)
		}
		for seat, want := range map[string]bool{EngineSeat: true, "node-login-shell": false} {
			verb := "acts"
			if !want {
				verb = "execute"
			}
			if heard, err := l.HeardServed(seat, verb); err != nil || heard != want {
				t.Errorf("discovery hears %s.%s here: %v, %v", seat, verb, heard, err)
			}
		}
	})
}